GH-1058: Fix Reject/Requeue for Async Returns

Fixes #1058

The listener adapter did not honor the exception types used to
complete the async result and always requeued.

Use trusty on travis.

Author: garyrussell

.travis.yml

@@ -1,3 +1,4 @@
+dist: trusty
 language: java
 jdk: oraclejdk8
 sudo: false

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java

@@ -396,7 +396,8 @@ public C createListenerContainer(RabbitListenerEndpoint endpoint) {
 				.acceptIfNotNull(this.beforeSendReplyPostProcessors, messageListener::setBeforeSendReplyPostProcessors)
 				.acceptIfNotNull(this.retryTemplate, messageListener::setRetryTemplate)
 				.acceptIfCondition(this.retryTemplate != null && this.recoveryCallback != null, this.recoveryCallback,
-					messageListener::setRecoveryCallback);
+					messageListener::setRecoveryCallback)
+				.acceptIfNotNull(this.defaultRequeueRejected, messageListener::setDefaultRequeueRejected);
 		}
 		initializeContainer(instance, endpoint);
 

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java

@@ -32,6 +32,7 @@
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.listener.ContainerUtils;
 import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
 import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
 import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
@@ -114,6 +115,7 @@ public abstract class AbstractAdaptableMessageListener implements ChannelAwareMe
 
 	private boolean isManualAck;
 
+	private boolean defaultRequeueRejected = true;
 
 	/**
 	 * Set the routing key to use when sending response messages.
@@ -260,6 +262,16 @@ protected MessageConverter getMessageConverter() {
 		return this.messageConverter;
 	}
 
+	/**
+	 * Set to the value of this listener's container equivalent property. Used when
+	 * rejecting from an async listener.
+	 * @param defaultRequeueRejected false to not requeue.
+	 * @since 2.1.8
+	 */
+	public void setDefaultRequeueRejected(boolean defaultRequeueRejected) {
+		this.defaultRequeueRejected = defaultRequeueRejected;
+	}
+
 	@Override
 	public void containerAckMode(AcknowledgeMode mode) {
 		this.isManualAck = AcknowledgeMode.MANUAL.equals(mode);
@@ -391,7 +403,8 @@ private void basicAck(Message request, Channel channel) {
 	private void asyncFailure(Message request, Channel channel, Throwable t) {
 		this.logger.error("Future or Mono was completed with an exception for " + request, t);
 		try {
-			channel.basicNack(request.getMessageProperties().getDeliveryTag(), false, true);
+			channel.basicNack(request.getMessageProperties().getDeliveryTag(), false,
+					ContainerUtils.shouldRequeue(this.defaultRequeueRejected, t, this.logger));
 		}
 		catch (IOException e) {
 			this.logger.error("Failed to nack message", e);

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AsyncListenerTests.java

@@ -20,7 +20,9 @@
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -29,6 +31,8 @@
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
+import org.springframework.amqp.AmqpRejectAndDontRequeueException;
+import org.springframework.amqp.ImmediateRequeueAmqpException;
 import org.springframework.amqp.core.AcknowledgeMode;
 import org.springframework.amqp.core.AnonymousQueue;
 import org.springframework.amqp.core.Queue;
@@ -88,6 +92,15 @@ public class AsyncListenerTests {
 	@Autowired
 	private Queue queue4;
 
+	@Autowired
+	private Queue queue5;
+
+	@Autowired
+	private Queue queue6;
+
+	@Autowired
+	private Queue queue7;
+
 	@Autowired
 	private Listener listener;
 
@@ -108,6 +121,19 @@ public void testAsyncListener() throws Exception {
 		assertThat(listener.latch4.await(10, TimeUnit.SECONDS));
 	}
 
+	@Test
+	public void testRouteToDLQ() throws InterruptedException {
+		this.rabbitTemplate.convertAndSend(this.queue5.getName(), "foo");
+		assertThat(this.listener.latch5.await(10, TimeUnit.SECONDS)).isTrue();
+		this.rabbitTemplate.convertAndSend(this.queue6.getName(), "foo");
+		assertThat(this.listener.latch6.await(10, TimeUnit.SECONDS)).isTrue();
+	}
+
+	@Test
+	public void testOverrideDontRequeue() throws Exception {
+		assertThat(this.rabbitTemplate.convertSendAndReceive(this.queue7.getName(), "foo")).isEqualTo("listen7");
+	}
+
 	@Configuration
 	@EnableRabbit
 	public static class EnableRabbitConfig {
@@ -131,6 +157,17 @@ public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
 			return factory;
 		}
 
+		@Bean
+		public SimpleRabbitListenerContainerFactory dontRequeueFactory() {
+			SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
+			factory.setConnectionFactory(rabbitConnectionFactory());
+			factory.setMismatchedQueuesFatal(true);
+			factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
+			factory.setMessageConverter(converter());
+			factory.setDefaultRequeueRejected(false);
+			return factory;
+		}
+
 		@Bean
 		public ConnectionFactory rabbitConnectionFactory() {
 			CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
@@ -180,6 +217,37 @@ public Queue queue4() {
 			return new AnonymousQueue();
 		}
 
+		@Bean
+		public Queue queue5() {
+			Map<String, Object> args = new HashMap<>();
+			args.put("x-dead-letter-exchange", "");
+			args.put("x-dead-letter-routing-key", queue5DLQ().getName());
+			return new AnonymousQueue(args);
+		}
+
+		@Bean
+		public Queue queue5DLQ() {
+			return new AnonymousQueue();
+		}
+
+		@Bean
+		public Queue queue6() {
+			Map<String, Object> args = new HashMap<>();
+			args.put("x-dead-letter-exchange", "");
+			args.put("x-dead-letter-routing-key", queue6DLQ().getName());
+			return new AnonymousQueue(args);
+		}
+
+		@Bean
+		public Queue queue6DLQ() {
+			return new AnonymousQueue();
+		}
+
+		@Bean
+		public Queue queue7() {
+			return new AnonymousQueue();
+		}
+
 		@Bean
 		public Listener listener() {
 			return new Listener();
@@ -196,6 +264,12 @@ public static class Listener {
 
 		private final CountDownLatch latch4 = new CountDownLatch(1);
 
+		private final CountDownLatch latch5 = new CountDownLatch(1);
+
+		private final CountDownLatch latch6 = new CountDownLatch(1);
+
+		private final AtomicBoolean first7 = new AtomicBoolean(true);
+
 		@RabbitListener(id = "foo", queues = "#{queue1.name}")
 		public ListenableFuture<String> listen1(String foo) {
 			SettableListenableFuture<String> future = new SettableListenableFuture<>();
@@ -231,6 +305,43 @@ public ListenableFuture<Void> listen4(@SuppressWarnings("unused") String foo) {
 			return future;
 		}
 
+		@RabbitListener(id = "fiz", queues = "#{queue5.name}")
+		public ListenableFuture<Void> listen5(@SuppressWarnings("unused") String foo) {
+			SettableListenableFuture<Void> future = new SettableListenableFuture<>();
+			future.setException(new AmqpRejectAndDontRequeueException("asyncToDLQ"));
+			return future;
+		}
+
+		@RabbitListener(id = "buz", queues = "#{queue5DLQ.name}")
+		public void listen5DLQ(@SuppressWarnings("unused") String foo) {
+			this.latch5.countDown();
+		}
+
+		@RabbitListener(id = "fix", queues = "#{queue6.name}", containerFactory = "dontRequeueFactory")
+		public ListenableFuture<Void> listen6(@SuppressWarnings("unused") String foo) {
+			SettableListenableFuture<Void> future = new SettableListenableFuture<>();
+			future.setException(new IllegalStateException("asyncDefaultToDLQ"));
+			return future;
+		}
+
+		@RabbitListener(id = "fox", queues = "#{queue6DLQ.name}")
+		public void listen6DLQ(@SuppressWarnings("unused") String foo) {
+			this.latch6.countDown();
+		}
+
+		@RabbitListener(id = "overrideFactoryRequeue", queues = "#{queue7.name}",
+				containerFactory = "dontRequeueFactory")
+		public ListenableFuture<String> listen7(@SuppressWarnings("unused") String foo) {
+			SettableListenableFuture<String> future = new SettableListenableFuture<>();
+			if (this.first7.compareAndSet(true, false)) {
+				future.setException(new ImmediateRequeueAmqpException("asyncOverrideDefaultToDLQ"));
+			}
+			else {
+				future.set("listen7");
+			}
+			return future;
+		}
+
 	}
 
 }

src/reference/asciidoc/amqp.adoc

@@ -2834,8 +2834,9 @@ Starting with version 2.1, `@RabbitListener` (and `@RabbitHandler`) methods can
 
 IMPORTANT: The listener container factory must be configured with `AcknowledgeMode.MANUAL` so that the consumer thread will not ack the message; instead, the asynchronous completion will ack or nack the message when the async operation completes.
 When the async result is completed with an error, whether the message is requeued or not depends on the exception type thrown, the container configuration, and the container error handler.
-By default, the message will be requeued, unless the container's `defaultRequeueRejected` property is set to `false`.
+By default, the message will be requeued, unless the container's `defaultRequeueRejected` property is set to `false` (it is `true` by default).
 If the async result is completed with an `AmqpRejectAndDontRequeueException`, the message will not be requeued.
+If the container's `defaultRequeueRejected` property is `false`, you can override that by setting the future's exception to a `ImmediateRequeueException` and the message will be requeued.
 If some exception occurs within the listener method that prevents creation of the async result object, you MUST catch that exception and return an appropriate return object that will cause the message to be acknowledged or requeued.
 
 [[threading]]

src/reference/asciidoc/whats-new.adoc

@@ -86,7 +86,7 @@ See <<async-annotation-driven-enable>> for more information.
 ===== Async `@RabbitListener` Return
 
 `@RabbitListener` methods can now return `ListenableFuture<?>` or `Mono<?>`.
-See <<async-return>> for more information.
+See <<async-returns>> for more information.
 
 ===== Connection Factory Bean Changes