GH-2920: SMLC: Interrupt ML on shutdown

Fixes: #2920

When the `SimpleMessageListenerContainer.shutdownTimeout` period is over,
there is no reason to keep any related threads active, sometimes blocking
the whole application stop.

* Introduce a `List SimpleMessageListenerContainer.processorThreadsToInterrupt`
to keep track of the scheduled `AsyncMessageProcessingConsumer`.
* Interrupt those threads when `this.cancellationLock.await()` is not successful
in the `SimpleMessageListenerContainer.shutdownAndWaitOrCallback()`

**Auto-cherry-pick to `3.2.x`**

Author: artembilan

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

@@ -115,6 +115,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 
 	private final ActiveObjectCounter<BlockingQueueConsumer> cancellationLock = new ActiveObjectCounter<>();
 
+	private final List<Thread> processorThreadsToInterrupt = new ArrayList<>();
+
 	private long startConsumerMinInterval = DEFAULT_START_CONSUMER_MIN_INTERVAL;
 
 	private long stopConsumerMinInterval = DEFAULT_STOP_CONSUMER_MIN_INTERVAL;
@@ -704,6 +706,7 @@ protected void shutdownAndWaitOrCallback(@Nullable Runnable callback) {
 				else {
 					logger.info("Workers not finished.");
 					if (isForceCloseChannel() || this.stopNow.get()) {
+						this.processorThreadsToInterrupt.forEach(Thread::interrupt);
 						canceledConsumers.forEach(consumer -> {
 							if (logger.isWarnEnabled()) {
 								logger.warn("Closing channel for unresponsive consumer: " + consumer);
@@ -1349,6 +1352,7 @@ public void run() { // NOSONAR - line count
 
 			try {
 				initialize();
+				SimpleMessageListenerContainer.this.processorThreadsToInterrupt.add(Thread.currentThread());
 				while (isActive(this.consumer) || this.consumer.hasDelivery() || !this.consumer.cancelled()) {
 					mainLoop();
 				}
@@ -1433,6 +1437,7 @@ public void run() { // NOSONAR - line count
 				}
 			}
 			finally {
+				SimpleMessageListenerContainer.this.processorThreadsToInterrupt.remove(Thread.currentThread());
 				SimpleMessageListenerContainer.this.cancellationLock.release(this.consumer);
 				if (getTransactionManager() != null) {
 					ConsumerChannelRegistry.unRegisterConsumerChannel();

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

@@ -40,6 +40,7 @@
 import com.rabbitmq.client.AMQP.BasicProperties;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Consumer;
+import com.rabbitmq.client.DefaultConsumer;
 import com.rabbitmq.client.Envelope;
 import com.rabbitmq.client.PossibleAuthenticationFailureException;
 import org.aopalliance.intercept.MethodInterceptor;
@@ -217,7 +218,7 @@ public void testTxSizeAcks() throws Exception {
 		container.setMessageListener(messages::add);
 		container.start();
 		BasicProperties props = new BasicProperties();
-		byte[] payload = "baz" .getBytes();
+		byte[] payload = "baz".getBytes();
 		Envelope envelope = new Envelope(1L, false, "foo", "bar");
 		consumer.get().handleDelivery("1", envelope, props, payload);
 		envelope = new Envelope(2L, false, "foo", "bar");
@@ -272,7 +273,7 @@ public void testTxSizeAcksWIthShortSet() throws Exception {
 		container.afterPropertiesSet();
 		container.start();
 		BasicProperties props = new BasicProperties();
-		byte[] payload = "baz" .getBytes();
+		byte[] payload = "baz".getBytes();
 		Envelope envelope = new Envelope(1L, false, "foo", "bar");
 		consumer.get().handleDelivery(consumerTag, envelope, props, payload);
 		envelope = new Envelope(2L, false, "foo", "bar");
@@ -756,6 +757,50 @@ void testShutdownWithPendingReplies() {
 		assertThat(replyCounter.getCount()).isEqualTo(1);
 	}
 
+	@Test
+	@SuppressWarnings("unchecked")
+	void listenerIsInterruptedOnUnsuccessfulShutdown() throws Exception {
+		ConnectionFactory connectionFactory = mock();
+		Connection connection = mock();
+		Channel channel = mock();
+		given(connectionFactory.createConnection()).willReturn(connection);
+		given(connection.createChannel(false)).willReturn(channel);
+		willAnswer(invocation -> "1")
+				.given(channel)
+				.basicConsume(anyString(), anyBoolean(), anyString(), anyBoolean(), anyBoolean(), anyMap(),
+						any(Consumer.class));
+		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
+		container.setQueueNames("some_queue");
+
+		CountDownLatch handlingLatch = new CountDownLatch(1);
+		CountDownLatch interruptedLatch = new CountDownLatch(1);
+		container.setMessageListener(message -> {
+			handlingLatch.countDown();
+			try {
+				Thread.sleep(2000L);
+			}
+			catch (InterruptedException e) {
+				interruptedLatch.countDown();
+			}
+		});
+		container.setShutdownTimeout(200L);
+		container.start();
+
+		Set<BlockingQueueConsumer> consumers = TestUtils.getPropertyValue(container, "consumers", Set.class);
+		BlockingQueueConsumer blockingQueueConsumer = consumers.iterator().next();
+
+		Map<String, DefaultConsumer> internalConsumers =
+				TestUtils.getPropertyValue(blockingQueueConsumer, "consumers", Map.class);
+		internalConsumers.values().iterator().next()
+				.handleDelivery("1", new Envelope(1, false, "", ""), new BasicProperties(), new byte[] {1});
+
+		assertThat(handlingLatch.await(10, TimeUnit.SECONDS)).isTrue();
+
+		container.stop();
+
+		assertThat(interruptedLatch.await(10, TimeUnit.SECONDS)).isTrue();
+	}
+
 	private Answer<Object> messageToConsumer(final Channel mockChannel, final SimpleMessageListenerContainer container,
 			final boolean cancel, final CountDownLatch latch) {
 		return invocation -> {