From b98847cd945eae3f0a50df4acdaefd55926ae58e Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 12 Jul 2023 14:56:48 -0400 Subject: [PATCH] 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 --- .../AbstractMessageListenerContainer.java | 41 ++++++++- .../listener/BlockingQueueConsumer.java | 18 ++-- .../DirectMessageListenerContainer.java | 89 +++++++++++++------ .../SimpleMessageListenerContainer.java | 29 +++--- ...sageListenerContainerIntegrationTests.java | 46 ++++++++++ ...ageListenerContainerIntegration2Tests.java | 44 ++++++++- src/reference/asciidoc/amqp.adoc | 16 ++++ 7 files changed, 227 insertions(+), 56 deletions(-) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java index 9011059a..65285217 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java +++ b/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 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 afterwards. * @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. diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java index e1359b3b..3d34f01d 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java +++ b/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 + } } /** diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java index a41fec74..09488e4d 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java +++ b/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 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 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 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 diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java index 4b4ca02c..daf95338 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java +++ b/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 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); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java index 71e1403d..62f111cc 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java +++ b/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(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java index d80eb17b..5599b99a 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java +++ b/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); diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 286ab70c..60e8e69b 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/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` 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 <> 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]]<> + +(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]]<> + (global-qos)