diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java index 75f6cc8ba..30d71550b 100644 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java +++ b/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; + } + } diff --git a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc index e9c94cb67..8c1923bec 100644 --- a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc +++ b/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 <> 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 <>. + Default: false transacted:: diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index be2035a22..3cb04d049 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/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 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); } diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index 69c1f21a6..bc2b2e37c 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/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 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