GH-159: Truncate stack traces if necessary

Fixes https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/159

spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java

@@ -109,6 +109,11 @@ public class RabbitConsumerProperties extends RabbitCommonProperties {
 	 */
 	private String consumerTagPrefix;
 
+	/**
+	 * Room to leave for other headers after adding the stack trace to a DLQ message.
+	 */
+	private int frameMaxHeadroom = 20_000;
+
 	public boolean isTransacted() {
 		return transacted;
 	}
@@ -259,4 +264,12 @@ public class RabbitConsumerProperties extends RabbitCommonProperties {
 		this.consumerTagPrefix = consumerTagPrefix;
 	}
 
+	public int getFrameMaxHeadroom() {
+		return this.frameMaxHeadroom;
+	}
+
+	public void setFrameMaxHeadroom(int frameMaxHeadroom) {
+		this.frameMaxHeadroom = frameMaxHeadroom;
+	}
+
 }

spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc

@@ -45,6 +45,7 @@ If retry is enabled (`maxAttempts > 1`), failed messages are delivered to the DL
 If retry is disabled (`maxAttempts = 1`), you should set `requeueRejected` to `false` (the default) so that failed messages are routed to the DLQ, instead of being re-queued.
 In addition, `republishToDlq` causes the binder to publish a failed message to the DLQ (instead of rejecting it).
 This feature lets additional information (such as the stack trace in the `x-exception-stacktrace` header) be added to the message in headers.
+See the <<spring-cloud-stream-rabbit-frame-max-headroom, `frameMaxHeadroom` property>> for information about truncated stack traces.
 This option does not need retry enabled.
 You can republish a failed message after just one attempt.
 Starting with version 1.2, you can configure the delivery mode of republished messages.
@@ -240,6 +241,14 @@ failedDeclarationRetryInterval::
 The interval (in milliseconds) between attempts to consume from a queue if it is missing.
 +
 Default: 5000
+[[spring-cloud-stream-rabbit-frame-max-headroom]]
+frameMaxHeadroom::
+The number of bytes to reserve for other headers when adding the stack trace to a DLQ message header.
+All headers must fit within the `frame_max` size configured on the broker.
+Stack traces can be large; if the size plus this property exceeds `frame_max` then the stack trace will be truncated.
+A WARN log will be written; consider increasing the `frame_max` or reducing the stack trace by catching the exception and throwing one with a smaller stack trace.
++
+Default: 20000
 headerPatterns::
 Patterns for headers to be mapped from inbound messages.
 +
@@ -312,6 +321,7 @@ republishToDlq::
 By default, messages that fail after retries are exhausted are rejected.
 If a dead-letter queue (DLQ) is configured, RabbitMQ routes the failed message (unchanged) to the DLQ.
 If set to `true`, the binder republishs failed messages to the DLQ with additional headers, including the exception message and stack trace from the cause of the final failure.
+Also see the <<spring-cloud-stream-rabbit-frame-max-headroom, frameMaxHeadroom property>>.
 +
 Default: false
 transacted::

spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java

@@ -31,6 +31,7 @@ import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.connection.LocalizedQueueConnectionFactory;
+import org.springframework.amqp.rabbit.connection.RabbitUtils;
 import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.core.support.BatchingStrategy;
@@ -489,6 +490,10 @@ public class RabbitMessageChannelBinder
 
 				private final String routingKey = properties.getExtension().getDeadLetterRoutingKey();
 
+				private final int frameMaxHeadroom = properties.getExtension().getFrameMaxHeadroom();
+
+				private int maxStackTraceLength = -1;
+
 				@Override
 				public void handleMessage(org.springframework.messaging.Message<?> message) throws MessagingException {
 					Message amqpMessage = (Message) message.getHeaders()
@@ -504,7 +509,21 @@ public class RabbitMessageChannelBinder
 						Throwable cause = (Throwable) message.getPayload();
 						MessageProperties messageProperties = amqpMessage.getMessageProperties();
 						Map<String, Object> headers = messageProperties.getHeaders();
-						headers.put(RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE, getStackTraceAsString(cause));
+						String stackTraceAsString = getStackTraceAsString(cause);
+						if (this.maxStackTraceLength < 0) {
+							int maxStackTraceLength = RabbitUtils
+									.getMaxFrame(this.template.getConnectionFactory());
+							if (maxStackTraceLength > 0) {
+								maxStackTraceLength -= this.frameMaxHeadroom;
+								this.maxStackTraceLength = maxStackTraceLength;
+							}
+						}
+						if (this.maxStackTraceLength > 0 && stackTraceAsString.length() > this.maxStackTraceLength) {
+							stackTraceAsString = stackTraceAsString.substring(0, this.maxStackTraceLength);
+							logger.warn("Stack trace in republished message header truncated due to frame_max limitations; "
+									+ "consider increasing frame_max on the broker or reduce the stack trace depth", cause);
+						}
+						headers.put(RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE, stackTraceAsString);
 						headers.put(RepublishMessageRecoverer.X_EXCEPTION_MESSAGE,
 								cause.getCause() != null ? cause.getCause().getMessage() : cause.getMessage());
 						headers.put(RepublishMessageRecoverer.X_ORIGINAL_EXCHANGE,
@@ -514,7 +533,7 @@ public class RabbitMessageChannelBinder
 						if (properties.getExtension().getRepublishDeliveyMode() != null) {
 							messageProperties.setDeliveryMode(properties.getExtension().getRepublishDeliveyMode());
 						}
-						template.send(this.exchange,
+						this.template.send(this.exchange,
 								this.routingKey != null ? this.routingKey : messageProperties.getConsumerQueue(),
 								amqpMessage);
 					}

spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java

@@ -16,6 +16,8 @@
 
 package org.springframework.cloud.stream.binder.rabbit;
 
+import java.io.PrintWriter;
+import java.io.StringWriter;
 import java.lang.reflect.Constructor;
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
@@ -46,12 +48,14 @@ import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.core.TopicExchange;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.connection.RabbitUtils;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitManagementTemplate;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.AsyncConsumerStartedEvent;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
 import org.springframework.amqp.support.AmqpHeaders;
 import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor;
 import org.springframework.amqp.utils.test.TestUtils;
@@ -103,6 +107,7 @@ import org.springframework.retry.support.RetryTemplate;
 import org.springframework.util.MimeTypeUtils;
 import org.springframework.util.ReflectionUtils;
 
+import com.rabbitmq.client.LongString;
 import com.rabbitmq.http.client.domain.QueueInfo;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -125,6 +130,10 @@ public class RabbitBinderTests extends
 
 	public static final String TEST_PREFIX = "bindertest.";
 
+	private static final String BIG_EXCEPTION_MESSAGE = new String(new byte[10_000]).replaceAll("\u0000", "x");
+
+	private int maxStackTraceSize;
+
 	@Rule
 	public RabbitTestSupport rabbitAvailableRule = new RabbitTestSupport(true);
 
@@ -1060,6 +1069,9 @@ public class RabbitBinderTests extends
 
 	@Test
 	public void testAutoBindDLQwithRepublish() throws Exception {
+		this.maxStackTraceSize = RabbitUtils.getMaxFrame(rabbitAvailableRule.getResource()) - 20_000;
+		assertThat(this.maxStackTraceSize).isGreaterThan(0);
+
 		RabbitTestBinder binder = getBinder();
 		ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties = createConsumerProperties();
 		consumerProperties.getExtension().setPrefix(TEST_PREFIX);
@@ -1069,11 +1081,13 @@ public class RabbitBinderTests extends
 		consumerProperties.getExtension().setDurableSubscription(true);
 		DirectChannel moduleInputChannel = createBindableChannel("input", createConsumerBindingProperties(consumerProperties));
 		moduleInputChannel.setBeanName("dlqPubTest");
+		RuntimeException exception = bigCause(new RuntimeException(BIG_EXCEPTION_MESSAGE));
+		assertThat(getStackTraceAsString(exception).length()).isGreaterThan(this.maxStackTraceSize);
 		moduleInputChannel.subscribe(new MessageHandler() {
 
 			@Override
 			public void handleMessage(Message<?> message) throws MessagingException {
-				throw new RuntimeException("foo");
+				throw exception;
 			}
 
 		});
@@ -1089,7 +1103,10 @@ public class RabbitBinderTests extends
 			org.springframework.amqp.core.Message deadLetter = template.receive(TEST_PREFIX + "foo.dlqpubtest.foo.dlq");
 			if (deadLetter != null) {
 				assertThat(new String(deadLetter.getBody())).isEqualTo("foo");
-				assertThat(deadLetter.getMessageProperties().getHeaders()).containsKey(("x-exception-stacktrace"));
+				assertThat(deadLetter.getMessageProperties().getHeaders())
+						.containsKey((RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE));
+				assertThat(((LongString) deadLetter.getMessageProperties().getHeaders()
+						.get(RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE)).length()).isEqualTo(this.maxStackTraceSize);
 				break;
 			}
 			Thread.sleep(100);
@@ -1625,6 +1642,20 @@ public class RabbitBinderTests extends
 		};
 	}
 
+	private RuntimeException bigCause(RuntimeException cause) {
+		if (getStackTraceAsString(cause).length() > this.maxStackTraceSize) {
+			return cause;
+		}
+		return bigCause(new RuntimeException(BIG_EXCEPTION_MESSAGE, cause));
+	}
+
+	private String getStackTraceAsString(Throwable cause) {
+		StringWriter stringWriter = new StringWriter();
+		PrintWriter printWriter = new PrintWriter(stringWriter, true);
+		cause.printStackTrace(printWriter);
+		return stringWriter.getBuffer().toString();
+	}
+
 	public static class TestPartitionKeyExtractorClass implements PartitionKeyExtractorStrategy {
 
 		@Override