From f8a67441b84d6a966edf9291937c48f309ce04c4 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 29 Mar 2017 15:37:15 -0400 Subject: [PATCH] GH-53: Add 'republishDeliveryMode' Resolves #53 Add a property to configure the delivery mode of republished DLQ messages. --- .../properties/RabbitConsumerProperties.java | 13 ++++++++- .../src/main/asciidoc/overview.adoc | 5 ++++ .../rabbit/RabbitMessageChannelBinder.java | 29 ++++++++++++++----- .../binder/rabbit/RabbitBinderTests.java | 4 +++ 4 files changed, 43 insertions(+), 8 deletions(-) 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 bdae3a24c..9b8ba62ef 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 @@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder.rabbit.properties; import javax.validation.constraints.Min; import org.springframework.amqp.core.AcknowledgeMode; +import org.springframework.amqp.core.MessageDeliveryMode; import org.springframework.util.Assert; /** @@ -41,6 +42,8 @@ public class RabbitConsumerProperties extends RabbitCommonProperties { private boolean republishToDlq; + private MessageDeliveryMode republishDeliveyMode = MessageDeliveryMode.PERSISTENT; + private boolean requeueRejected = false; private String[] headerPatterns = new String[] {"*"}; @@ -60,7 +63,7 @@ public class RabbitConsumerProperties extends RabbitCommonProperties { } public void setAcknowledgeMode(AcknowledgeMode acknowledgeMode) { - Assert.notNull("Acknowledge mode cannot be null"); + Assert.notNull(acknowledgeMode, "Acknowledge mode cannot be null"); this.acknowledgeMode = acknowledgeMode; } @@ -129,6 +132,14 @@ public class RabbitConsumerProperties extends RabbitCommonProperties { return requeueRejected; } + public MessageDeliveryMode getRepublishDeliveyMode() { + return this.republishDeliveyMode; + } + + public void setRepublishDeliveyMode(MessageDeliveryMode republishDeliveyMode) { + this.republishDeliveyMode = republishDeliveyMode; + } + public void setRequeueRejected(boolean requeueRejected) { this.requeueRejected = requeueRejected; } 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 7bfa8447c..f35b79e4a 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 @@ -44,6 +44,7 @@ If retry is enabled (`maxAttempts > 1`) failed messages will be delivered to the If retry is disabled (`maxAttempts = 1`), you should set `requeueRejected` to `false` (default) so that a failed message will be routed to the DLQ, instead of being requeued. In addition, `republishToDlq` causes the binder to publish a failed message to the DLQ (instead of rejecting it); this enables additional information to be added to the message in headers, such as the stack trace in the `x-exception-stacktrace` header. 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 messsages; see property `republishDeliveryMode`. IMPORTANT: Setting `requeueRejected` to `true` will cause the message to be requeued and redelivered continually, which is likely not what you want unless the failure issue is transient. In general, it's better to enable retry within the binder by setting `maxAttempts` to greater than one, or set `republishToDlq` to `true`. @@ -220,6 +221,10 @@ requeueRejected:: Whether delivery failures should be requeued when retry is disabled or republishToDlq is false. + Default: `false`. +republishDeliveryMode:: + When `republishToDlq` is `true`, specify the delivery mode of the republished message. ++ +Default: `DeliveryMode.PERSISTENT` republishToDlq:: By default, messages which fail after retries are exhausted are rejected. If a dead-letter queue (DLQ) is configured, RabbitMQ will route the failed message (unchanged) to the DLQ. 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 b11906a87..3d8bbd0f1 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 @@ -20,6 +20,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.List; +import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder; @@ -259,8 +260,7 @@ public class RabbitMessageChannelBinder if (properties.getMaxAttempts() > 1 || properties.getExtension().isRepublishToDlq()) { RetryOperationsInterceptor retryInterceptor = RetryInterceptorBuilder.stateless() .retryOperations(buildRetryTemplate(properties)) - .recoverer(determineRecoverer(baseQueueName, properties.getExtension(), - properties.getExtension().isRepublishToDlq())) + .recoverer(determineRecoverer(baseQueueName, properties.getExtension())) .build(); listenerContainer.setAdviceChain(retryInterceptor); } @@ -293,12 +293,27 @@ public class RabbitMessageChannelBinder provisioningProvider.cleanAutoDeclareContext(consumerDestination.getName()); } - private MessageRecoverer determineRecoverer(String name, RabbitCommonProperties properties, boolean republish) { - if (republish) { + private MessageRecoverer determineRecoverer(String name, final RabbitConsumerProperties properties) { + if (properties.isRepublishToDlq()) { RabbitTemplate errorTemplate = new RabbitTemplate(this.connectionFactory); - return new RepublishMessageRecoverer(errorTemplate, - deadLetterExchangeName(properties), - applyPrefix(properties.getPrefix(), name)); + if (properties.getRepublishDeliveyMode() != null) { + return new RepublishMessageRecoverer(errorTemplate, + deadLetterExchangeName(properties), + applyPrefix(properties.getPrefix(), name)) { + + @Override + public void recover(Message message, Throwable cause) { + message.getMessageProperties().setDeliveryMode(properties.getRepublishDeliveyMode()); + super.recover(message, cause); + } + + }; + } + else { + return new RepublishMessageRecoverer(errorTemplate, + deadLetterExchangeName(properties), + applyPrefix(properties.getPrefix(), name)); + } } else { return new RejectAndDontRequeueRecoverer(); 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 1755fecbd..5e81a33f2 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 @@ -636,6 +636,7 @@ public class RabbitBinderTests extends properties.getExtension().setPrefix("bindertest."); properties.getExtension().setAutoBindDlq(true); properties.getExtension().setRepublishToDlq(true); + properties.getExtension().setRepublishDeliveyMode(MessageDeliveryMode.NON_PERSISTENT); properties.setMaxAttempts(1); // disable retry properties.setPartitioned(true); properties.setInstanceIndex(0); @@ -706,6 +707,8 @@ public class RabbitBinderTests extends assertThat(received.getMessageProperties().getHeaders().get("x-original-routingKey")) .isEqualTo("partPubDLQ.0-1"); assertThat(received.getMessageProperties().getHeaders()).doesNotContainKey(BinderHeaders.PARTITION_HEADER); + assertThat(received.getMessageProperties().getReceivedDeliveryMode()) + .isEqualTo(MessageDeliveryMode.NON_PERSISTENT); output.send(new GenericMessage<>(0)); received = template.receive(streamDLQName); @@ -800,6 +803,7 @@ public class RabbitBinderTests extends assertThat(received.getMessageProperties().getReceivedRoutingKey()) .isEqualTo("bindertest.partDLQ.1.dlqPartGrp-1"); assertThat(received.getMessageProperties().getHeaders()).doesNotContainKey(BinderHeaders.PARTITION_HEADER); + assertThat(received.getMessageProperties().getReceivedDeliveryMode()).isEqualTo(MessageDeliveryMode.PERSISTENT); output.send(new GenericMessage(0)); received = template.receive(streamDLQName);