GH-2482: Fix Async Container Stop

- Stop with callback did not reset `active` flag (both containers)
- DMLC did not release the `cancellationLock`
- Also restore `@LongRunning`

(cherry picked from commit d44a8d0)

Author: garyrussell

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

@@ -1350,9 +1350,19 @@ public void initialize() {
 	}
 
 	/**
-	 * Stop the shared Connection, call {@link #doShutdown()}, and close this container.
+	 * Stop the shared Connection, call {@link #shutdown(Runnable)}, and close this
+	 * container.
 	 */
 	public void shutdown() {
+		shutdown(null);
+	}
+
+	/**
+	 * Stop the shared Connection, call {@link #shutdownAndWaitOrCallback(Runnable)}, and
+	 * close this container.
+	 * @param callback an optional {@link Runnable} to call when the stop is complete.
+	 */
+	public void shutdown(@Nullable Runnable callback) {
 		synchronized (this.lifecycleMonitor) {
 			if (!isActive()) {
 				logger.debug("Shutdown ignored - container is not active already");
@@ -1367,7 +1377,7 @@ public void shutdown() {
 
 		// Shut down the invokers.
 		try {
-			doShutdown();
+			shutdownAndWaitOrCallback(callback);
 		}
 		catch (Exception ex) {
 			throw convertRabbitAccessException(ex);
@@ -1405,10 +1415,7 @@ protected void doShutdown() {
 
 	@Override
 	public void stop(Runnable callback) {
-		shutdownAndWaitOrCallback(() -> {
-			setNotRunning();
-			callback.run();
-		});
+		shutdown(callback);
 	}
 
 	protected void shutdownAndWaitOrCallback(@Nullable Runnable callback) {

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

@@ -1337,14 +1337,14 @@ void cancelConsumer(final String eventMessage) {
 
 		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);
 		}
 
 		@Override

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

@@ -801,7 +801,7 @@ public void onMessage(Message message) {
 	}
 
 	@Test
-	void forceStop() {
+	void forceStop() throws InterruptedException {
 		CountDownLatch latch1 = new CountDownLatch(1);
 		CachingConnectionFactory cf = new CachingConnectionFactory("localhost");
 		DirectMessageListenerContainer container = new DirectMessageListenerContainer(cf);
@@ -812,6 +812,7 @@ void forceStop() {
 		try {
 			container.setQueueNames(Q3);
 			container.setForceStop(true);
+			container.setShutdownTimeout(20_000L);
 			template.convertAndSend(Q3, "one");
 			template.convertAndSend(Q3, "two");
 			template.convertAndSend(Q3, "three");
@@ -828,14 +829,21 @@ void forceStop() {
 				assertThat(queueInfo).isNotNull();
 				assertThat(queueInfo.getMessageCount()).isEqualTo(0);
 			});
+			CountDownLatch latch2 = new CountDownLatch(1);
+			long t1 = System.currentTimeMillis();
 			container.stop(() -> {
+				latch2.countDown();
 			});
 			latch1.countDown();
+			assertThat(System.currentTimeMillis() - t1).isLessThan(5_000L);
 			await().untilAsserted(() -> {
 				QueueInformation queueInfo = admin.getQueueInfo(Q3);
 				assertThat(queueInfo).isNotNull();
 				assertThat(queueInfo.getMessageCount()).isEqualTo(4);
 			});
+			assertThat(latch2.await(10, TimeUnit.SECONDS)).isTrue();
+			assertThat(container.isActive()).isFalse();
+			assertThat(container.isRunning()).isFalse();
 		}
 		finally {
 			container.stop();

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

@@ -67,6 +67,7 @@
 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 +96,7 @@
  */
 @RabbitAvailable(queues = { SimpleMessageListenerContainerIntegration2Tests.TEST_QUEUE,
 		SimpleMessageListenerContainerIntegration2Tests.TEST_QUEUE_1 })
-//@LongRunning
+@LongRunning
 public class SimpleMessageListenerContainerIntegration2Tests {
 
 	public static final String TEST_QUEUE = "test.queue.SimpleMessageListenerContainerIntegration2Tests";

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

@@ -437,6 +437,7 @@ public void testCallbackIsRunOnStopAlsoWhenNoConsumerIsActive() throws Interrupt
 		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
 
 		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
+		ReflectionTestUtils.setField(container, "active", Boolean.TRUE);
 
 		final CountDownLatch countDownLatch = new CountDownLatch(1);
 		container.stop(countDownLatch::countDown);
@@ -449,6 +450,7 @@ public void testCallbackIsRunOnStopAlsoWhenContainerIsStoppingForAbort() throws
 
 		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
 		ReflectionTestUtils.setField(container, "containerStoppingForAbort", new AtomicReference<>(new Thread()));
+		ReflectionTestUtils.setField(container, "active", Boolean.TRUE);
 
 		final CountDownLatch countDownLatch = new CountDownLatch(1);
 		container.stop(countDownLatch::countDown);