GH-2941: Fix BlockingQueueConsumer race condition on cancel

Fixes: #2941
Issue link: #2941

Now `BlockingQueueConsumer.basicCancel()` performs `RabbitUtils.closeMessageConsumer()`
to initiate `basicRecovery` on the transactional consumer to re-queue all the un-acked messages.
However, there is a race condition when one in-flight message may still be delivered
to the listener and then TX commit is initiated.
There a `basicAck()` is initiated.
However, such a tag might already be discarded because of the previous `basicRecovery`.
Therefore, adjust `BlockingQueueConsumer.commitIfNecessary()` to skip `basicAck()` if locally transacted and already cancelled.
Right, this may lead to the duplication delivery, but having abnormal shutdown situation we cannot guarantee that this message
to commit has been processed properly.
Also, adjust `BlockingQueueConsumer.nextMessage()` to rollback a message if consumer is canceled instead of
going through the loop via listener

* Increase `replyTimeout` in the `EnableRabbitReturnTypesTests` for resource-sensitive builds

Author: artembilan

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

@@ -170,7 +170,8 @@ public class BlockingQueueConsumer {
 
 	private long consumeDelay;
 
-	private java.util.function.Consumer<String> missingQueuePublisher = str -> { };
+	private java.util.function.Consumer<String> missingQueuePublisher = str -> {
+	};
 
 	private boolean globalQos;
 
@@ -468,12 +469,13 @@ protected void basicCancel() {
 
 	protected void basicCancel(boolean expected) {
 		this.normalCancel = expected;
+		this.cancelled.set(true);
+		this.abortStarted = System.currentTimeMillis();
+
 		Collection<String> consumerTags = getConsumerTags();
 		if (!CollectionUtils.isEmpty(consumerTags)) {
 			RabbitUtils.closeMessageConsumer(this.channel, consumerTags, this.transactional);
 		}
-		this.cancelled.set(true);
-		this.abortStarted = System.currentTimeMillis();
 	}
 
 	protected boolean hasDelivery() {
@@ -560,12 +562,26 @@ public Message nextMessage(long timeout) throws InterruptedException, ShutdownSi
 		if (!this.missingQueues.isEmpty()) {
 			checkMissingQueues();
 		}
-		Message message = handle(this.queue.poll(timeout, TimeUnit.MILLISECONDS));
-		if (message == null && this.cancelled.get()) {
+		if (!cancelled()) {
+			Message message = handle(this.queue.poll(timeout, TimeUnit.MILLISECONDS));
+			if (message != null && cancelled()) {
+				this.activeObjectCounter.release(this);
+				ConsumerCancelledException consumerCancelledException = new ConsumerCancelledException();
+				rollbackOnExceptionIfNecessary(consumerCancelledException,
+						message.getMessageProperties().getDeliveryTag());
+				throw consumerCancelledException;
+			}
+			else {
+				return message;
+			}
+		}
+		else {
+			this.deliveryTags.clear();
 			this.activeObjectCounter.release(this);
-			throw new ConsumerCancelledException();
+			ConsumerCancelledException consumerCancelledException = new ConsumerCancelledException();
+			rollbackOnExceptionIfNecessary(consumerCancelledException);
+			throw consumerCancelledException;
 		}
-		return message;
 	}
 
 	/*
@@ -792,7 +808,7 @@ public void stop() {
 			if (this.abortStarted == 0) { // signal handle delivery to use offer
 				this.abortStarted = System.currentTimeMillis();
 			}
-			if (!this.cancelled()) {
+			if (!cancelled()) {
 				try {
 					RabbitUtils.closeMessageConsumer(this.channel, getConsumerTags(), this.transactional);
 				}
@@ -894,22 +910,25 @@ boolean commitIfNecessary(boolean localTx, boolean forceAck) {
 		/*
 		 * If we have a TX Manager, but no TX, act like we are locally transacted.
 		 */
-		boolean isLocallyTransacted = localTx
-				|| (this.transactional
-				&& TransactionSynchronizationManager.getResource(this.connectionFactory) == null);
+		boolean isLocallyTransacted =
+				localTx ||
+						(this.transactional &&
+								TransactionSynchronizationManager.getResource(this.connectionFactory) == null);
 		try {
 			boolean ackRequired = forceAck || (!this.acknowledgeMode.isAutoAck() && !this.acknowledgeMode.isManual());
 
-			if (ackRequired && (!this.transactional || isLocallyTransacted)) {
-				long deliveryTag = new ArrayList<>(this.deliveryTags).get(this.deliveryTags.size() - 1);
-				try {
-					this.channel.basicAck(deliveryTag, true);
-					notifyMessageAckListener(true, deliveryTag, null);
-				}
-				catch (Exception e) {
-					logger.error("Error acking.", e);
-					notifyMessageAckListener(false, deliveryTag, e);
-				}
+			if (ackRequired && (!this.transactional || (isLocallyTransacted && !cancelled()))) {
+				OptionalLong deliveryTag = this.deliveryTags.stream().mapToLong(l -> l).max();
+				deliveryTag.ifPresent((tag) -> {
+					try {
+						this.channel.basicAck(tag, true);
+						notifyMessageAckListener(true, tag, null);
+					}
+					catch (Exception e) {
+						logger.error("Error acking.", e);
+						notifyMessageAckListener(false, tag, e);
+					}
+				});
 			}
 
 			if (isLocallyTransacted) {

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

@@ -113,6 +113,7 @@ public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(Cachi
 		public RabbitTemplate template(CachingConnectionFactory cf, Jackson2JsonMessageConverter converter) {
 			RabbitTemplate template = new RabbitTemplate(cf);
 			template.setMessageConverter(converter);
+			template.setReplyTimeout(30_000);
 			return template;
 		}