From ffbda2e852b319979b9415584732dffce17f9b06 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 28 Jan 2020 09:35:53 -0500 Subject: [PATCH] GH-281: Republish to DLQ ack original when MANUAL Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/281 When using `AcknowledgeMode.MANUAL` with `republishToDlq`, original message is left in an un-ack'd state. Always ack the original message after republishing to the DLQ. --- .../rabbit/RabbitMessageChannelBinder.java | 26 ++++++- .../binder/rabbit/RabbitBinderTests.java | 75 +++++++++++++++++++ 2 files changed, 97 insertions(+), 4 deletions(-) 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 65297eca5..134ab194d 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 @@ -16,6 +16,7 @@ package org.springframework.cloud.stream.binder.rabbit; +import java.io.IOException; import java.io.PrintWriter; import java.io.StringWriter; import java.util.ArrayList; @@ -26,10 +27,12 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.regex.Pattern; import com.rabbitmq.client.AMQP; +import com.rabbitmq.client.Channel; import com.rabbitmq.client.Envelope; import org.springframework.amqp.AmqpRejectAndDontRequeueException; import org.springframework.amqp.ImmediateAcknowledgeAmqpException; +import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.core.MessageProperties; @@ -49,6 +52,7 @@ import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer; import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter; import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException; import org.springframework.amqp.rabbit.support.MessagePropertiesConverter; +import org.springframework.amqp.support.AmqpHeaders; import org.springframework.amqp.support.converter.AbstractMessageConverter; import org.springframework.amqp.support.converter.MessageConversionException; import org.springframework.amqp.support.converter.SimpleMessageConverter; @@ -588,6 +592,7 @@ public class RabbitMessageChannelBinder extends protected MessageHandler getErrorMessageHandler(ConsumerDestination destination, String group, final ExtendedConsumerProperties properties) { + if (properties.getExtension().isRepublishToDlq()) { return new MessageHandler() { @@ -610,9 +615,7 @@ public class RabbitMessageChannelBinder extends private int maxStackTraceLength = -1; @Override - public void handleMessage( - org.springframework.messaging.Message message) - throws MessagingException { + public void handleMessage(org.springframework.messaging.Message message) throws MessagingException { Message amqpMessage = StaticMessageHeaderAccessor.getSourceData(message); if (!(message instanceof ErrorMessage)) { @@ -669,6 +672,21 @@ public class RabbitMessageChannelBinder extends this.routingKey != null ? this.routingKey : messageProperties.getConsumerQueue(), amqpMessage); + if (properties.getExtension().getAcknowledgeMode().equals(AcknowledgeMode.MANUAL)) { + org.springframework.messaging.Message original = + ((ErrorMessage) message).getOriginalMessage(); + if (original != null) { + // If we are using manual acks, ack the original message. + try { + original.getHeaders().get(AmqpHeaders.CHANNEL, Channel.class) + .basicAck(original.getHeaders() + .get(AmqpHeaders.DELIVERY_TAG, Long.class), false); + } + catch (IOException e) { + logger.debug("Failed to ack original message", e); + } + } + } } } @@ -713,7 +731,7 @@ public class RabbitMessageChannelBinder extends + message.getClass().toString() + " for: " + message); throw new ListenerExecutionFailedException( "Unexpected error message " + message, - new AmqpRejectAndDontRequeueException(""), null); + new AmqpRejectAndDontRequeueException(""), (Message[]) null); } else if (amqpMessage == null) { logger.error("No raw message header in " + message); 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 b9ad84fe3..bf113f869 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 @@ -976,6 +976,81 @@ public class RabbitBinderTests extends assertThat(context.containsBean(TEST_PREFIX + "dlqtest.default.dlq")).isFalse(); } + @Test + public void testAutoBindDLQManualAcks() throws Exception { + RabbitTestBinder binder = getBinder(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setPrefix(TEST_PREFIX); + consumerProperties.getExtension().setAutoBindDlq(true); + consumerProperties.setMaxAttempts(2); + consumerProperties.getExtension().setDurableSubscription(true); + consumerProperties.getExtension().setAcknowledgeMode(AcknowledgeMode.MANUAL); + BindingProperties bindingProperties = createConsumerBindingProperties( + consumerProperties); + DirectChannel moduleInputChannel = createBindableChannel("input", + bindingProperties); + moduleInputChannel.setBeanName("dlqTestManual"); + Client client = new Client("http://guest:guest@localhost:15672/api"); + moduleInputChannel.subscribe(new MessageHandler() { + + @Override + public void handleMessage(Message message) throws MessagingException { + // Wait until the unacked state is reflected in the admin + QueueInfo info = client.getQueue("/", TEST_PREFIX + "dlqTestManual.default"); + int n = 0; + while (n++ < 100 && info.getMessagesUnacknowledged() < 1L) { + try { + Thread.sleep(100); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + info = client.getQueue("/", TEST_PREFIX + "dlqTestManual.default"); + } + throw new RuntimeException("foo"); + } + + }); + Binding consumerBinding = binder.bindConsumer("dlqTestManual", + "default", moduleInputChannel, consumerProperties); + + RabbitTemplate template = new RabbitTemplate( + this.rabbitAvailableRule.getResource()); + template.convertAndSend("", TEST_PREFIX + "dlqTestManual.default", "foo"); + + int n = 0; + while (n++ < 100) { + Object deadLetter = template + .receiveAndConvert(TEST_PREFIX + "dlqTestManual.default.dlq"); + if (deadLetter != null) { + assertThat(deadLetter).isEqualTo("foo"); + break; + } + Thread.sleep(100); + } + assertThat(n).isLessThan(100); + + n = 0; + QueueInfo info = client.getQueue("/", TEST_PREFIX + "dlqTestManual.default"); + while (n++ < 100 && info.getMessagesUnacknowledged() > 0L) { + Thread.sleep(100); + info = client.getQueue("/", TEST_PREFIX + "dlqTestManual.default"); + } + assertThat(info.getMessagesUnacknowledged()).isEqualTo(0L); + + consumerBinding.unbind(); + + ApplicationContext context = TestUtils.getPropertyValue(binder, + "binder.provisioningProvider.autoDeclareContext", + ApplicationContext.class); + assertThat(context.containsBean(TEST_PREFIX + "dlqTestManual.default.binding")) + .isFalse(); + assertThat(context.containsBean(TEST_PREFIX + "dlqTestManual.default")).isFalse(); + assertThat(context.containsBean(TEST_PREFIX + "dlqTestManual.default.dlq.binding")) + .isFalse(); + assertThat(context.containsBean(TEST_PREFIX + "dlqTestManual.default.dlq")).isFalse(); + } + @Test public void testAutoBindDLQPartionedConsumerFirst() throws Exception { RabbitTestBinder binder = getBinder();