GH-2482: Option for Containers to Stop Immediately

Resolves https://github.com/spring-projects/spring-amqp/issues/2482

`forceStop` means stop after the current record and requeue all prefetched.

Just close the channel - canceling the consumer first causes a race condition which
could allow another exclusive or single-active consumer to start processing while
this container is still running.

Also support async stop on DMLC (previously only available on the SMLC).

**cherry-pick to 2.4.x**
# Conflicts:
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java

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

@@ -141,9 +141,9 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 
 	private final Map<String, String> micrometerTags = new HashMap<>();
 
-	private ContainerDelegate proxy = this.delegate;
+	protected final AtomicBoolean stopNow = new AtomicBoolean(); // NOSONAR
 
-	private final AtomicBoolean logDeclarationException = new AtomicBoolean(true);
+	private ContainerDelegate proxy = this.delegate;
 
 	private long shutdownTimeout = DEFAULT_SHUTDOWN_TIMEOUT;
 
@@ -258,6 +258,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 
 	private MessageAckListener messageAckListener = (success, deliveryTag, cause) -> { };
 
+	private boolean forceStop;
+
 	@Override
 	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
 		this.applicationEventPublisher = applicationEventPublisher;
@@ -1212,6 +1214,25 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 		return this.messageAckListener;
 	}
 
+	/**
+	 * Stop container after current message(s) are processed and requeue any prefetched.
+	 * @return true to stop when current message(s) are processed.
+	 * @since 2.4.14
+	 */
+	protected boolean isForceStop() {
+		return this.forceStop;
+	}
+
+	/**
+	 * Set to true to stop the container after the current message(s) are processed and
+	 * requeue any prefetched. Useful when using exclusive or single-active consumers.
+	 * @param forceStop true to stop when current messsage(s) are processed.
+	 * @since 2.4.14
+	 */
+	public void setForceStop(boolean forceStop) {
+		this.forceStop = forceStop;
+	}
+
 	/**
 	 * Delegates to {@link #validateConfiguration()} and {@link #initialize()}.
 	 */
@@ -1376,7 +1397,21 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 	 * A shared Rabbit Connection, if any, will automatically be closed <i>afterwards</i>.
 	 * @see #shutdown()
 	 */
-	protected abstract void doShutdown();
+	protected void doShutdown() {
+		shutdownAndWaitOrCallback(null);
+	}
+
+	@Override
+	public void stop(Runnable callback) {
+		shutdownAndWaitOrCallback(() -> {
+			setNotRunning();
+			callback.run();
+		});
+	}
+
+	protected void shutdownAndWaitOrCallback(@Nullable Runnable callback) {
+	}
+
 
 	/**
 	 * @return Whether this container is currently active, that is, whether it has been set up but not shut down yet.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -784,11 +784,17 @@ public class BlockingQueueConsumer {
 		if (logger.isDebugEnabled()) {
 			logger.debug("Closing Rabbit Channel: " + this.channel);
 		}
-		RabbitUtils.setPhysicalCloseRequired(this.channel, true);
-		ConnectionFactoryUtils.releaseResources(this.resourceHolder);
-		this.deliveryTags.clear();
-		this.consumers.clear();
-		this.queue.clear(); // in case we still have a client thread blocked
+		forceCloseAndClearQueue();
+	}
+
+	public void forceCloseAndClearQueue() {
+		if (this.channel != null && this.channel.isOpen()) {
+			RabbitUtils.setPhysicalCloseRequired(this.channel, true);
+			ConnectionFactoryUtils.releaseResources(this.resourceHolder);
+			this.deliveryTags.clear();
+			this.consumers.clear();
+			this.queue.clear(); // in case we still have a client thread blocked
+		}
 	}
 
 	/**

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2021 the original author or authors.
+ * Copyright 2016-2023 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.
@@ -812,7 +812,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 	}
 
 	@Override
-	protected void doShutdown() {
+	protected void shutdownAndWaitOrCallback(@Nullable Runnable callback) {
 		LinkedList<SimpleConsumer> canceledConsumers = null;
 		boolean waitForConsumers = false;
 		synchronized (this.consumersMonitor) {
@@ -825,36 +825,53 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 			}
 		}
 		if (waitForConsumers) {
-			try {
-				if (this.cancellationLock.await(getShutdownTimeout(), TimeUnit.MILLISECONDS)) {
-					this.logger.info("Successfully waited for consumers to finish.");
-				}
-				else {
-					this.logger.info("Consumers not finished.");
-					if (isForceCloseChannel()) {
-						canceledConsumers.forEach(consumer -> {
-							String eventMessage = "Closing channel for unresponsive consumer: " + consumer;
-							if (logger.isWarnEnabled()) {
-								logger.warn(eventMessage);
-							}
-							consumer.cancelConsumer(eventMessage);
-						});
+			LinkedList<SimpleConsumer> consumersToWait = canceledConsumers;
+			Runnable awaitShutdown = () -> {
+				try {
+					if (this.cancellationLock.await(getShutdownTimeout(), TimeUnit.MILLISECONDS)) {
+						this.logger.info("Successfully waited for consumers to finish.");
+					}
+					else {
+						this.logger.info("Consumers not finished.");
+						if (isForceCloseChannel() || this.stopNow.get()) {
+							consumersToWait.forEach(consumer -> {
+								String eventMessage = "Closing channel for unresponsive consumer: " + consumer;
+								if (logger.isWarnEnabled()) {
+									logger.warn(eventMessage);
+								}
+								consumer.cancelConsumer(eventMessage);
+							});
+						}
 					}
 				}
+				catch (InterruptedException e) {
+					Thread.currentThread().interrupt();
+					this.logger.warn("Interrupted waiting for consumers. Continuing with shutdown.");
+				}
+				finally {
+					this.startedLatch = new CountDownLatch(1);
+					this.started = false;
+					this.aborted = false;
+					this.hasStopped = true;
+				}
+				this.stopNow.set(false);
+				runCallbackIfNotNull(callback);
+			};
+			if (callback == null) {
+				awaitShutdown.run();
 			}
-			catch (InterruptedException e) {
-				Thread.currentThread().interrupt();
-				this.logger.warn("Interrupted waiting for consumers. Continuing with shutdown.");
-			}
-			finally {
-				this.startedLatch = new CountDownLatch(1);
-				this.started = false;
-				this.aborted = false;
-				this.hasStopped = true;
+			else {
+				getTaskExecutor().execute(awaitShutdown);
 			}
 		}
 	}
 
+	private void runCallbackIfNotNull(@Nullable Runnable callback) {
+		if (callback != null) {
+			callback.run();
+		}
+	}
+
 	/**
 	 * Must hold this.consumersMonitor.
 	 * @param consumers a copy of this.consumers.
@@ -862,7 +879,12 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 	private void actualShutDown(List<SimpleConsumer> consumers) {
 		Assert.state(getTaskExecutor() != null, "Cannot shut down if not initialized");
 		this.logger.debug("Shutting down");
-		consumers.forEach(this::cancelConsumer);
+		if (isForceStop()) {
+			this.stopNow.set(true);
+		}
+		else {
+			consumers.forEach(this::cancelConsumer);
+		}
 		this.consumers.clear();
 		this.consumersByQueue.clear();
 		this.logger.debug("All consumers canceled");
@@ -1030,6 +1052,10 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 		public void handleDelivery(String consumerTag, Envelope envelope,
 				BasicProperties properties, byte[] body) {
 
+			if (!getChannel().isOpen()) {
+				this.logger.debug("Discarding prefetch, channel closed");
+				return;
+			}
 			MessageProperties messageProperties =
 					getMessagePropertiesConverter().toMessageProperties(properties, envelope, "UTF-8");
 			messageProperties.setConsumerTag(consumerTag);
@@ -1071,6 +1097,9 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 					// NOSONAR
 				}
 			}
+			if (DirectMessageListenerContainer.this.stopNow.get()) {
+				closeChannel();
+			}
 		}
 
 		private void executeListenerInTransaction(Object data, long deliveryTag) {
@@ -1307,11 +1336,15 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 		}
 
 		private void finalizeConsumer() {
+			closeChannel();
+			DirectMessageListenerContainer.this.cancellationLock.release(this);
+			consumerRemoved(this);
+		}
+
+		private void closeChannel() {
 			RabbitUtils.setPhysicalCloseRequired(getChannel(), true);
 			RabbitUtils.closeChannel(getChannel());
 			RabbitUtils.closeConnection(this.connection);
-			DirectMessageListenerContainer.this.cancellationLock.release(this);
-			consumerRemoved(this);
 		}
 
 		@Override

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

@@ -607,19 +607,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 	}
 
 	@Override
-	protected void doShutdown() {
-		shutdownAndWaitOrCallback(null);
-	}
-
-	@Override
-	public void stop(Runnable callback) {
-		shutdownAndWaitOrCallback(() -> {
-			setNotRunning();
-			callback.run();
-		});
-	}
-
-	private void shutdownAndWaitOrCallback(@Nullable Runnable callback) {
+	protected void shutdownAndWaitOrCallback(@Nullable Runnable callback) {
 		Thread thread = this.containerStoppingForAbort.get();
 		if (thread != null && !thread.equals(Thread.currentThread())) {
 			logger.info("Shutdown ignored - container is stopping due to an aborted consumer");
@@ -631,9 +619,14 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 		synchronized (this.consumersMonitor) {
 			if (this.consumers != null) {
 				Iterator<BlockingQueueConsumer> consumerIterator = this.consumers.iterator();
+				if (isForceStop()) {
+					this.stopNow.set(true);
+				}
 				while (consumerIterator.hasNext()) {
 					BlockingQueueConsumer consumer = consumerIterator.next();
-					consumer.basicCancel(true);
+					if (!isForceStop()) {
+						consumer.basicCancel(true);
+					}
 					canceledConsumers.add(consumer);
 					consumerIterator.remove();
 					if (consumer.declaring) {
@@ -657,7 +650,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 				}
 				else {
 					logger.info("Workers not finished.");
-					if (isForceCloseChannel()) {
+					if (isForceCloseChannel() || this.stopNow.get()) {
 						canceledConsumers.forEach(consumer -> {
 							if (logger.isWarnEnabled()) {
 								logger.warn("Closing channel for unresponsive consumer: " + consumer);
@@ -676,7 +669,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 				this.consumers = null;
 				this.cancellationLock.deactivate();
 			}
-
+			this.stopNow.set(false);
 			runCallbackIfNotNull(callback);
 		};
 		if (callback == null) {
@@ -1323,6 +1316,10 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 
 		private void mainLoop() throws Exception { // NOSONAR Exception
 			try {
+				if (SimpleMessageListenerContainer.this.stopNow.get()) {
+					this.consumer.forceCloseAndClearQueue();
+					return;
+				}
 				boolean receivedOk = receiveAndExecute(this.consumer); // At least one message received
 				if (SimpleMessageListenerContainer.this.maxConcurrentConsumers != null) {
 					checkAdjust(receivedOk);

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java

@@ -51,6 +51,7 @@ import org.springframework.amqp.AmqpAuthenticationException;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageListener;
 import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.QueueInformation;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.connection.Connection;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
@@ -90,6 +91,7 @@ import com.rabbitmq.client.Consumer;
  */
 @RabbitAvailable(queues = { DirectMessageListenerContainerIntegrationTests.Q1,
 		DirectMessageListenerContainerIntegrationTests.Q2,
+		DirectMessageListenerContainerIntegrationTests.Q3,
 		DirectMessageListenerContainerIntegrationTests.EQ1,
 		DirectMessageListenerContainerIntegrationTests.EQ2,
 		DirectMessageListenerContainerIntegrationTests.DLQ1 })
@@ -102,6 +104,8 @@ public class DirectMessageListenerContainerIntegrationTests {
 
 	public static final String Q2 = "testQ2.DirectMessageListenerContainerIntegrationTests";
 
+	public static final String Q3 = "testQ3.DirectMessageListenerContainerIntegrationTests";
+
 	public static final String EQ1 = "eventTestQ1.DirectMessageListenerContainerIntegrationTests";
 
 	public static final String EQ2 = "eventTestQ2.DirectMessageListenerContainerIntegrationTests";
@@ -792,6 +796,48 @@ public class DirectMessageListenerContainerIntegrationTests {
 		assertThat(ackDeliveryTag.get()).isEqualTo(1);
 	}
 
+	@Test
+	void forceStop() {
+		CountDownLatch latch1 = new CountDownLatch(1);
+		CachingConnectionFactory cf = new CachingConnectionFactory("localhost");
+		DirectMessageListenerContainer container = new DirectMessageListenerContainer(cf);
+		container.setMessageListener((ChannelAwareMessageListener) (msg, chan) -> {
+			latch1.await(10, TimeUnit.SECONDS);
+		});
+		RabbitTemplate template = new RabbitTemplate(cf);
+		try {
+			container.setQueueNames(Q3);
+			container.setForceStop(true);
+			template.convertAndSend(Q3, "one");
+			template.convertAndSend(Q3, "two");
+			template.convertAndSend(Q3, "three");
+			template.convertAndSend(Q3, "four");
+			template.convertAndSend(Q3, "five");
+			await().untilAsserted(() -> {
+				QueueInformation queueInfo = admin.getQueueInfo(Q3);
+				assertThat(queueInfo).isNotNull();
+				assertThat(queueInfo.getMessageCount()).isEqualTo(5);
+			});
+			container.start();
+			await().untilAsserted(() -> {
+				QueueInformation queueInfo = admin.getQueueInfo(Q3);
+				assertThat(queueInfo).isNotNull();
+				assertThat(queueInfo.getMessageCount()).isEqualTo(0);
+			});
+			container.stop(() -> {
+			});
+			latch1.countDown();
+			await().untilAsserted(() -> {
+				QueueInformation queueInfo = admin.getQueueInfo(Q3);
+				assertThat(queueInfo).isNotNull();
+				assertThat(queueInfo.getMessageCount()).isEqualTo(4);
+			});
+		}
+		finally {
+			container.stop();
+		}
+	}
+
 	@Test
 	public void testMessageAckListenerWithBatchAck() throws Exception {
 		final AtomicInteger calledTimes = new AtomicInteger();

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -57,6 +57,7 @@ import org.springframework.amqp.core.BatchMessageListener;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageListener;
 import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.QueueInformation;
 import org.springframework.amqp.event.AmqpEvent;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.connection.Connection;
@@ -66,7 +67,6 @@ import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.junit.BrokerTestUtils;
-import org.springframework.amqp.rabbit.junit.LongRunning;
 import org.springframework.amqp.rabbit.junit.RabbitAvailable;
 import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
 import org.springframework.amqp.rabbit.listener.adapter.ReplyingMessageListener;
@@ -95,7 +95,7 @@ import com.rabbitmq.client.Channel;
  */
 @RabbitAvailable(queues = { SimpleMessageListenerContainerIntegration2Tests.TEST_QUEUE,
 		SimpleMessageListenerContainerIntegration2Tests.TEST_QUEUE_1 })
-@LongRunning
+//@LongRunning
 public class SimpleMessageListenerContainerIntegration2Tests {
 
 	public static final String TEST_QUEUE = "test.queue.SimpleMessageListenerContainerIntegration2Tests";
@@ -747,6 +747,44 @@ public class SimpleMessageListenerContainerIntegration2Tests {
 		assertThat(ackDeliveryTag.get()).isEqualTo(messageCount);
 	}
 
+	@Test
+	void forceStop() {
+		CountDownLatch latch1 = new CountDownLatch(1);
+		this.container = createContainer((ChannelAwareMessageListener) (msg, chan) -> {
+			latch1.await(10, TimeUnit.SECONDS);
+		}, false, TEST_QUEUE);
+		try {
+			this.container.setForceStop(true);
+			this.template.convertAndSend(TEST_QUEUE, "one");
+			this.template.convertAndSend(TEST_QUEUE, "two");
+			this.template.convertAndSend(TEST_QUEUE, "three");
+			this.template.convertAndSend(TEST_QUEUE, "four");
+			this.template.convertAndSend(TEST_QUEUE, "five");
+			await().untilAsserted(() -> {
+				QueueInformation queueInfo = admin.getQueueInfo(TEST_QUEUE);
+				assertThat(queueInfo).isNotNull();
+				assertThat(queueInfo.getMessageCount()).isEqualTo(5);
+			});
+			this.container.start();
+			await().untilAsserted(() -> {
+				QueueInformation queueInfo = admin.getQueueInfo(TEST_QUEUE);
+				assertThat(queueInfo).isNotNull();
+				assertThat(queueInfo.getMessageCount()).isEqualTo(0);
+			});
+			this.container.stop(() -> {
+			});
+			latch1.countDown();
+			await().untilAsserted(() -> {
+				QueueInformation queueInfo = admin.getQueueInfo(TEST_QUEUE);
+				assertThat(queueInfo).isNotNull();
+				assertThat(queueInfo.getMessageCount()).isEqualTo(4);
+			});
+		}
+		finally {
+			this.container.stop();
+		}
+	}
+
 	private boolean containerStoppedForAbortWithBadListener() throws InterruptedException {
 		Log logger = spy(TestUtils.getPropertyValue(container, "logger", Log.class));
 		new DirectFieldAccessor(container).setPropertyValue("logger", logger);

src/reference/asciidoc/amqp.adoc

@@ -3467,6 +3467,10 @@ Starting with version 1.5, you can now assign a `group` to the container on the
 This provides a mechanism to get a reference to a subset of containers.
 Adding a `group` attribute causes a bean of type `Collection<MessageListenerContainer>` to be registered with the context with the group name.
 
+By default, stopping a container will cancel the consumer and process all prefetched messages before stopping.
+Starting with versions 2.4.14, 3.0.6, you can set the <<forceStop>> container property to true to stop immediately after the current message is processed, causing any prefetched messages to be requeued.
+This is useful, for example, if exclusive or single-active consumers are being used.
+
 [[receiving-batch]]
 ===== @RabbitListener with Batching
 
@@ -6199,6 +6203,18 @@ a|image::images/tickmark.png[]
 a|image::images/tickmark.png[]
 a|
 
+|[[forceStop]]<<forceStop,`forceStop`>> +
+(N/A)
+
+|Set to true to stop (when the container is stopped) after the current record is processed; causing all prefetched messages to be requeued.
+By default, the container will cancel the consumer and process all prefetched messages before stopping.
+Since versions 2.4.14, 3.0.6
+Defaults to `false`.
+
+a|image::images/tickmark.png[]
+a|image::images/tickmark.png[]
+a|
+
 |[[globalQos]]<<globalQos,`globalQos`>> +
 (global-qos)