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.

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<RabbitConsumerProperties> 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);

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<RabbitConsumerProperties> 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<MessageChannel> 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();