GH-1409: Fix Nacks for Async Replies

Resolves #1409

Normally, when message has a fatal exception (such as message conversion)
`basicNack` with `multiple` true is used, to nack any previously unacked
messages (e.g. when using batch size to limit the ack traffic).
Even when using manual acks, fatal exceptions are nacked by the container
because the user does not have access to the message.

However, when using async replies, this has the side effect of nacking
unprocessed messages.

Detect whether async replies are being used and only nack individual
records that cause fatal exceptions.

Also, coerce the `AcknowledgeMode` to `MANUAL` for such listners.

Add a test for both containers; send a good message followed by a
bad one without actually completing the reply future.
After the exception occurs and the container is stopped, there should
be one messag in the queue.

* Remove warning, deprecation; add docs.

* Docs.

**Cherry-pick to `2.3.x` & `2.2.x`**
# Conflicts:
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java

# Conflicts:
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/DelegatingInvocableHandler.java
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/HandlerAdapter.java

Author: garyrussell

spring-amqp/src/main/java/org/springframework/amqp/core/MessageListener.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -43,6 +43,16 @@ default void containerAckMode(AcknowledgeMode mode) {
 		// NOSONAR - empty
 	}
 
+	/**
+	 * Return true if this listener is request/reply and the replies are
+	 * async.
+	 * @return true for async replies.
+	 * @since 2.2.21
+	 */
+	default boolean isAsyncReplies() {
+		return false;
+	}
+
 	/**
 	 * Delivers a batch of messages.
 	 * @param messages the messages.

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

@@ -250,6 +250,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 
 	private volatile boolean lazyLoad;
 
+	private boolean asyncReplies;
+
 	@Override
 	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
 		this.applicationEventPublisher = applicationEventPublisher;
@@ -433,6 +435,7 @@ public void setMessageListener(MessageListener messageListener) {
 		this.messageListener = messageListener;
 		this.isBatchListener = messageListener instanceof BatchMessageListener
 				|| messageListener instanceof ChannelAwareBatchMessageListener;
+		this.asyncReplies = messageListener.isAsyncReplies();
 	}
 
 	/**
@@ -1009,10 +1012,12 @@ public boolean isPossibleAuthenticationFailureFatal() {
 		return this.possibleAuthenticationFailureFatal;
 	}
 
-
 	protected boolean isPossibleAuthenticationFailureFatalSet() {
 		return this.possibleAuthenticationFailureFatalSet;
 	}
+	protected boolean isAsyncReplies() {
+		return this.asyncReplies;
+	}
 
 	/**
 	 * Set to true to automatically declare elements (queues, exchanges, bindings)
@@ -1193,6 +1198,9 @@ public final void afterPropertiesSet() {
 		catch (IllegalStateException e) {
 			this.logger.debug("Could not enable micrometer timers", e);
 		}
+		if (this.isAsyncReplies() && !AcknowledgeMode.MANUAL.equals(this.acknowledgeMode)) {
+			this.acknowledgeMode = AcknowledgeMode.MANUAL;
+		}
 	}
 
 	@Override

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

@@ -758,6 +758,17 @@ public synchronized void stop() {
 	 * @param ex the thrown application exception or error
 	 */
 	public void rollbackOnExceptionIfNecessary(Throwable ex) {
+		rollbackOnExceptionIfNecessary(ex, -1);
+	}
+
+	/**
+	 * Perform a rollback, handling rollback exceptions properly.
+	 * @param ex the thrown application exception or error
+	 * @param tag delivery tag; when specified (greater than or equal to 0) only that
+	 * message is nacked.
+	 * @since 2.2.21.
+	 */
+	public void rollbackOnExceptionIfNecessary(Throwable ex, long tag) {
 
 		boolean ackRequired = !this.acknowledgeMode.isAutoAck()
 				&& (!this.acknowledgeMode.isManual() || ContainerUtils.isRejectManual(ex));
@@ -769,14 +780,20 @@ public void rollbackOnExceptionIfNecessary(Throwable ex) {
 				RabbitUtils.rollbackIfNecessary(this.channel);
 			}
 			if (ackRequired) {
-				OptionalLong deliveryTag = this.deliveryTags.stream().mapToLong(l -> l).max();
-				if (deliveryTag.isPresent()) {
-					this.channel.basicNack(deliveryTag.getAsLong(), true,
-							ContainerUtils.shouldRequeue(this.defaultRequeueRejected, ex, logger));
+				if (tag < 0) {
+					OptionalLong deliveryTag = this.deliveryTags.stream().mapToLong(l -> l).max();
+					if (deliveryTag.isPresent()) {
+						this.channel.basicNack(deliveryTag.getAsLong(), true,
+								ContainerUtils.shouldRequeue(this.defaultRequeueRejected, ex, logger));
+					}
+					if (this.transactional) {
+						// Need to commit the reject (=nack)
+						RabbitUtils.commitIfNecessary(this.channel);
+					}
 				}
-				if (this.transactional) {
-					// Need to commit the reject (=nack)
-					RabbitUtils.commitIfNecessary(this.channel);
+				else {
+					this.channel.basicNack(tag, false,
+							ContainerUtils.shouldRequeue(this.defaultRequeueRejected, ex, logger));
 				}
 			}
 		}
@@ -785,7 +802,12 @@ public void rollbackOnExceptionIfNecessary(Throwable ex) {
 			throw RabbitExceptionTranslator.convertRabbitAccessException(e); // NOSONAR stack trace loss
 		}
 		finally {
-			this.deliveryTags.clear();
+			if (tag < 0) {
+				this.deliveryTags.clear();
+			}
+			else {
+				this.deliveryTags.remove(tag);
+			}
 		}
 	}
 

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

@@ -1196,7 +1196,7 @@ private void rollback(long deliveryTag, Exception e) {
 							}
 						}
 					}
-					getChannel().basicNack(deliveryTag, true,
+					getChannel().basicNack(deliveryTag, !isAsyncReplies(),
 							ContainerUtils.shouldRequeue(isDefaultRequeueRejected(), e, this.logger));
 				}
 				catch (IOException e1) {

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

@@ -991,6 +991,9 @@ private boolean doReceiveAndExecute(BlockingQueueConsumer consumer) throws Excep
 						}
 						break;
 					}
+					long tagToRollback = isAsyncReplies()
+							? message.getMessageProperties().getDeliveryTag()
+							: -1;
 					if (getTransactionManager() != null) {
 						if (getTransactionAttribute().rollbackOn(ex)) {
 							RabbitResourceHolder resourceHolder = (RabbitResourceHolder) TransactionSynchronizationManager
@@ -1003,7 +1006,7 @@ private boolean doReceiveAndExecute(BlockingQueueConsumer consumer) throws Excep
 								 * If we don't actually have a transaction, we have to roll back
 								 * manually. See prepareHolderForRollback().
 								 */
-								consumer.rollbackOnExceptionIfNecessary(ex);
+								consumer.rollbackOnExceptionIfNecessary(ex, tagToRollback);
 							}
 							throw ex; // encompassing transaction will handle the rollback.
 						}
@@ -1015,7 +1018,7 @@ private boolean doReceiveAndExecute(BlockingQueueConsumer consumer) throws Excep
 						}
 					}
 					else {
-						consumer.rollbackOnExceptionIfNecessary(ex);
+						consumer.rollbackOnExceptionIfNecessary(ex, tagToRollback);
 						throw ex;
 					}
 				}

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

@@ -21,7 +21,6 @@
 import java.lang.reflect.Type;
 import java.lang.reflect.WildcardType;
 import java.util.Arrays;
-import java.util.function.Consumer;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -56,7 +55,6 @@
 import org.springframework.util.concurrent.ListenableFuture;
 
 import com.rabbitmq.client.Channel;
-import reactor.core.publisher.Mono;
 
 /**
  * An abstract {@link org.springframework.amqp.core.MessageListener} adapter providing the
@@ -81,7 +79,7 @@ public abstract class AbstractAdaptableMessageListener implements ChannelAwareMe
 
 	private static final ParserContext PARSER_CONTEXT = new TemplateParserContext("!{", "}");
 
-	private static final boolean monoPresent = // NOSONAR - lower case
+	static final boolean monoPresent = // NOSONAR - lower case, protected
 			ClassUtils.isPresent("reactor.core.publisher.Mono", ChannelAwareMessageListener.class.getClassLoader());
 
 	/**
@@ -635,19 +633,4 @@ public Object getResult() {
 
 	}
 
-	private static class MonoHandler { // NOSONAR - pointless to name it ..Utils|Helper
-
-		static boolean isMono(Object result) {
-			return result instanceof Mono;
-		}
-
-		@SuppressWarnings("unchecked")
-		static void subscribe(Object returnValue, Consumer<? super Object> success,
-				Consumer<? super Throwable> failure, Runnable completeConsumer) {
-
-			((Mono<? super Object>) returnValue).subscribe(success, failure, completeConsumer);
-		}
-
-	}
-
 }

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

@@ -20,6 +20,7 @@
 import java.lang.reflect.Method;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -41,6 +42,7 @@
 import org.springframework.messaging.handler.annotation.SendTo;
 import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
 import org.springframework.util.Assert;
+import org.springframework.util.concurrent.ListenableFuture;
 
 
 /**
@@ -75,6 +77,8 @@ public class DelegatingInvocableHandler {
 
 	private final BeanExpressionContext beanExpressionContext;
 
+	private final boolean asyncReplies;
+
 	/**
 	 * Construct an instance with the supplied handlers for the bean.
 	 * @param handlers the handlers.
@@ -106,6 +110,18 @@ public DelegatingInvocableHandler(List<InvocableHandlerMethod> handlers,
 		this.bean = bean;
 		this.resolver = beanExpressionResolver;
 		this.beanExpressionContext = beanExpressionContext;
+		boolean asyncReplies;
+		asyncReplies = defaultHandler != null && isAsyncReply(defaultHandler);
+		Iterator<InvocableHandlerMethod> iterator = handlers.iterator();
+		while (iterator.hasNext()) {
+			asyncReplies |= isAsyncReply(iterator.next());
+		}
+		this.asyncReplies = asyncReplies;
+	}
+
+	private boolean isAsyncReply(InvocableHandlerMethod method) {
+		return (AbstractAdaptableMessageListener.monoPresent && MonoHandler.isMono(method.getMethod().getReturnType()))
+				|| ListenableFuture.class.isAssignableFrom(method.getMethod().getReturnType());
 	}
 
 	/**
@@ -115,6 +131,15 @@ public Object getBean() {
 		return this.bean;
 	}
 
+	/**
+	 * Return true if any handler method has an async reply type.
+	 * @return the asyncReply.
+	 * @since 2.2.21
+	 */
+	public boolean isAsyncReplies() {
+		return this.asyncReplies;
+	}
+
 	/**
 	 * Invoke the method with the given message.
 	 * @param message the message.

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

@@ -22,6 +22,7 @@
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
+import org.springframework.util.concurrent.ListenableFuture;
 
 /**
  * A wrapper for either an {@link InvocableHandlerMethod} or
@@ -38,14 +39,24 @@ public class HandlerAdapter {
 
 	private final DelegatingInvocableHandler delegatingHandler;
 
+	private final boolean asyncReplies;
+
+	/**
+	 * Construct an instance with the provided method.
+	 * @param invokerHandlerMethod the method.
+	 */
 	public HandlerAdapter(InvocableHandlerMethod invokerHandlerMethod) {
 		this.invokerHandlerMethod = invokerHandlerMethod;
 		this.delegatingHandler = null;
+		this.asyncReplies = (AbstractAdaptableMessageListener.monoPresent
+				&& MonoHandler.isMono(invokerHandlerMethod.getMethod().getReturnType()))
+			|| ListenableFuture.class.isAssignableFrom(invokerHandlerMethod.getMethod().getReturnType());
 	}
 
 	public HandlerAdapter(DelegatingInvocableHandler delegatingHandler) {
 		this.invokerHandlerMethod = null;
 		this.delegatingHandler = delegatingHandler;
+		this.asyncReplies = delegatingHandler.isAsyncReplies();
 	}
 
 	public InvocationResult invoke(Message<?> message, Object... providedArgs) throws Exception { // NOSONAR
@@ -126,6 +137,22 @@ public Object getBean() {
 		}
 	}
 
+	/**
+	 * Return true if any handler method has an async reply type.
+	 * @return the asyncReply.
+	 * @since 2.2.21
+	 */
+	public boolean isAsyncReplies() {
+		return this.asyncReplies;
+	}
+
+	/**
+	 * Build an {@link InvocationResult} for the result and inbound payload.
+	 * @param result the result.
+	 * @param inboundPayload the payload.
+	 * @return the invocation result.
+	 * @since 2.1.7
+	 */
 	@Nullable
 	public InvocationResult getInvocationResultFor(Object result, Object inboundPayload) {
 		if (this.invokerHandlerMethod != null) {

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

@@ -103,6 +103,11 @@ public void setHandlerAdapter(HandlerAdapter handlerAdapter) {
 		this.handlerAdapter = handlerAdapter;
 	}
 
+	@Override
+	public boolean isAsyncReplies() {
+		return this.handlerAdapter.isAsyncReplies();
+	}
+
 	/**
 	 * Set the {@link AmqpHeaderMapper} implementation to use to map the standard
 	 * AMQP headers. By default, a {@link org.springframework.amqp.support.SimpleAmqpHeaderMapper

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

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2021 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.amqp.rabbit.listener.adapter;
+
+import java.util.function.Consumer;
+
+import reactor.core.publisher.Mono;
+
+/**
+ * Class to prevent direct links to {@link Mono}.
+ * @author Gary Russell
+ * @since 2.2.21
+ */
+final class MonoHandler { // NOSONAR - pointless to name it ..Utils|Helper
+
+	private MonoHandler() {
+	}
+
+	static boolean isMono(Object result) {
+		return result instanceof Mono;
+	}
+
+	@SuppressWarnings("unchecked")
+	static void subscribe(Object returnValue, Consumer<? super Object> success,
+			Consumer<? super Throwable> failure, Runnable completeConsumer) {
+
+		((Mono<? super Object>) returnValue).subscribe(success, failure, completeConsumer);
+	}
+
+}