From a69d2309eb47ba8890aaf4b88ef63f5f07235e99 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 9 Oct 2018 13:08:37 -0400 Subject: [PATCH] GH-180: republishToDlq and ImmediateAckAmqpEx Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/180 --- .../src/main/asciidoc/overview.adoc | 3 ++ .../rabbit/RabbitMessageChannelBinder.java | 23 +++++++++ .../binder/rabbit/RabbitBinderTests.java | 47 +++++++++---------- 3 files changed, 48 insertions(+), 25 deletions(-) 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 8c1923bec..64c461ae2 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 @@ -51,6 +51,9 @@ You can republish a failed message after just one attempt. Starting with version 1.2, you can configure the delivery mode of republished messages. See the <>. +If the stream listener throws an `ImmediateAcknowledgeAmqpException`, the DLQ is bypassed and the message simply discarded. +Starting with version 2.1, this is true regardless of the setting of `republishToDlq`; previously it was only the case when `republishToDlq` was `false`. + IMPORTANT: Setting `requeueRejected` to `true` (with `republishToDlq=false` ) causes the message to be re-queued and redelivered continually, which is likely not what you want unless the reason for the failure is transient. In general, you should enable retry within the binder by setting `maxAttempts` to greater than one or by setting `republishToDlq` to `true`. 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 4465394f8..cf6e16416 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 @@ -25,6 +25,7 @@ import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; import org.springframework.amqp.AmqpRejectAndDontRequeueException; +import org.springframework.amqp.ImmediateAcknowledgeAmqpException; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.core.MessageProperties; @@ -507,6 +508,12 @@ public class RabbitMessageChannelBinder } else { Throwable cause = (Throwable) message.getPayload(); + if (!shouldRepublish(cause)) { + if (logger.isDebugEnabled()) { + logger.debug("Skipping republish of: " + message); + } + return; + } MessageProperties messageProperties = amqpMessage.getMessageProperties(); Map headers = messageProperties.getHeaders(); String stackTraceAsString = getStackTraceAsString(cause); @@ -539,6 +546,22 @@ public class RabbitMessageChannelBinder } } + /** + * Traverse the cause tree, stopping at AmqpRejectAndDontRequeueException + * or ImmediateAcknowledgeAmqpException. + * @param throwable the throwable. + * @return true if neither found or AmqpRejectAndDontRequeueException is + * found first. + */ + private boolean shouldRepublish(Throwable throwable) { + Throwable cause = throwable; + while (cause != null && !(cause instanceof AmqpRejectAndDontRequeueException) + && !(cause instanceof ImmediateAcknowledgeAmqpException)) { + cause = cause.getCause(); + } + return !(cause instanceof ImmediateAcknowledgeAmqpException); + } + }; } else if (properties.getMaxAttempts() > 1) { 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 bc2b2e37c..952fac654 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 @@ -36,6 +36,7 @@ import org.junit.rules.TestName; import org.mockito.ArgumentCaptor; import org.springframework.amqp.AmqpIOException; +import org.springframework.amqp.ImmediateAcknowledgeAmqpException; import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.AnonymousQueue; @@ -1083,10 +1084,14 @@ public class RabbitBinderTests extends moduleInputChannel.setBeanName("dlqPubTest"); RuntimeException exception = bigCause(new RuntimeException(BIG_EXCEPTION_MESSAGE)); assertThat(getStackTraceAsString(exception).length()).isGreaterThan(this.maxStackTraceSize); + AtomicBoolean dontRepublish = new AtomicBoolean(); moduleInputChannel.subscribe(new MessageHandler() { @Override public void handleMessage(Message message) throws MessagingException { + if (dontRepublish.get()) { + throw new ImmediateAcknowledgeAmqpException("testDontRepublish"); + } throw exception; } @@ -1098,34 +1103,26 @@ public class RabbitBinderTests extends RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource()); template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtest.foo", "foo"); - int n = 0; - while (n++ < 100) { - 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((RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE)); - assertThat(((LongString) deadLetter.getMessageProperties().getHeaders() - .get(RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE)).length()).isEqualTo(this.maxStackTraceSize); - break; - } - Thread.sleep(100); - } - assertThat(n).isLessThan(100); + template.setReceiveTimeout(10_000); + org.springframework.amqp.core.Message deadLetter = template.receive(TEST_PREFIX + "foo.dlqpubtest.foo.dlq"); + assertThat(deadLetter).isNotNull(); + assertThat(new String(deadLetter.getBody())).isEqualTo("foo"); + assertThat(deadLetter.getMessageProperties().getHeaders()) + .containsKey((RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE)); + assertThat(((LongString) deadLetter.getMessageProperties().getHeaders() + .get(RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE)).length()).isEqualTo(this.maxStackTraceSize); template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtest2.foo", "bar"); - n = 0; - while (n++ < 100) { - org.springframework.amqp.core.Message deadLetter = template.receive(TEST_PREFIX + "foo.dlqpubtest2.foo.dlq"); - if (deadLetter != null) { - assertThat(new String(deadLetter.getBody())).isEqualTo("bar"); - assertThat(deadLetter.getMessageProperties().getHeaders()).containsKey(("x-exception-stacktrace")); - break; - } - Thread.sleep(100); - } - assertThat(n).isLessThan(100); + deadLetter = template.receive(TEST_PREFIX + "foo.dlqpubtest2.foo.dlq"); + assertThat(deadLetter).isNotNull(); + assertThat(new String(deadLetter.getBody())).isEqualTo("bar"); + assertThat(deadLetter.getMessageProperties().getHeaders()).containsKey(("x-exception-stacktrace")); + + dontRepublish.set(true); + template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtest2.foo", "baz"); + template.setReceiveTimeout(500); + assertThat(template.receive(TEST_PREFIX + "foo.dlqpubtest2.foo.dlq")).isNull(); consumerBinding.unbind(); }