From 80db8b184ee9c5ca7280ea43eccaca811b8b9544 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 21 Nov 2017 15:57:41 -0500 Subject: [PATCH] AMQP-785: SMLC Lifecycle fixes JIRA: https://jira.spring.io/browse/AMQP-785 Fixes: spring-projects/spring-amqp#689 - Only stop the container on one thread - Ignore concurrent stops - Interrupt consumer threads that are attempting to declare queues - In `restart()` don't start a new consumer if the container is stopping - Defer publishing consumer failure events until container is stopped - Add a RecoveryListener if needed to ensure channels are never recovered - Fix event publishing for `Error` - it is fatal __backport to 1.7.x will require work__ Clear the declaring flag when exiting `start()` with exception. Release the `activeObjectCounter` when interrupted while declaring. Polishing stopped container lifecycle Since restarted consumer is not be aware about stopped container, it can restart properly when RabbitMQ comes back on-line independently of the container state * Add `active` flag to the `ActiveObjectCounter` and `deactivate()` and `isActive()` hooks * Use `ActiveObjectCounter.deactivate()` in the container shutdown * Use `ActiveObjectCounter.isActive()` in the `BlockingQueueConsumer.cancelled()` * Use `BlockingQueueConsumer.cancelled()` in its `start()` toi check container activity before performing network job * Check `isActive()` state in the `AbstractMessageListenerContainer.shutdown()` * Remove `SimpleMessageListenerContainer.containerStopping` in favor of `isActive()` hook Conflicts: build.gradle spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ActiveObjectCounter.java spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java Resolved. --- build.gradle | 4 +- .../AbstractMessageListenerContainer.java | 10 ++- .../rabbit/listener/ActiveObjectCounter.java | 15 ++++ .../listener/BlockingQueueConsumer.java | 72 +++++++++++++++-- .../SimpleMessageListenerContainer.java | 78 +++++++++++++++---- 5 files changed, 156 insertions(+), 23 deletions(-) diff --git a/build.gradle b/build.gradle index 299dce05..1b142286 100644 --- a/build.gradle +++ b/build.gradle @@ -103,9 +103,9 @@ subprojects { subproject -> rabbitmqHttpClientVersion = '1.1.1.RELEASE' slf4jVersion = "1.7.25" - springVersion = project.hasProperty('springVersion') ? project.springVersion : '4.3.11.RELEASE' + springVersion = project.hasProperty('springVersion') ? project.springVersion : '4.3.13.RELEASE' - springRetryVersion = '1.2.0.RELEASE' + springRetryVersion = '1.2.1.RELEASE' } eclipse { 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 37b7d84f..0ea118cc 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 @@ -547,12 +547,17 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor * Stop the shared Connection, call {@link #doShutdown()}, and close this container. */ public void shutdown() { - logger.debug("Shutting down Rabbit listener container"); synchronized (this.lifecycleMonitor) { + if (!isActive()) { + logger.info("Shutdown ignored - container is not active already"); + return; + } this.active = false; this.lifecycleMonitor.notifyAll(); } + logger.debug("Shutting down Rabbit listener container"); + // Shut down the invokers. try { doShutdown(); @@ -602,6 +607,9 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor */ @Override public void start() { + if (isRunning()) { + return; + } if (!this.initialized) { synchronized (this.lifecycleMonitor) { if (!this.initialized) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ActiveObjectCounter.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ActiveObjectCounter.java index 10689dd9..e0d341fb 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ActiveObjectCounter.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ActiveObjectCounter.java @@ -24,13 +24,19 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; /** + * A mechanism to keep track of active objects. + * @param the object type. + * * @author Dave Syer + * @author Artem Bilan * */ public class ActiveObjectCounter { private final ConcurrentMap locks = new ConcurrentHashMap(); + private volatile boolean active = true; + public void add(T object) { CountDownLatch lock = new CountDownLatch(1); this.locks.putIfAbsent(object, lock); @@ -71,6 +77,15 @@ public class ActiveObjectCounter { public void reset() { this.locks.clear(); + this.active = true; + } + + public void deactivate() { + this.active = false; + } + + public boolean isActive() { + return this.active; } } 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 a38ee502..3a1d315a 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 @@ -64,7 +64,10 @@ import com.rabbitmq.client.AlreadyClosedException; import com.rabbitmq.client.Channel; import com.rabbitmq.client.DefaultConsumer; import com.rabbitmq.client.Envelope; +import com.rabbitmq.client.Recoverable; +import com.rabbitmq.client.RecoveryListener; import com.rabbitmq.client.ShutdownSignalException; +import com.rabbitmq.client.impl.recovery.AutorecoveringChannel; import com.rabbitmq.utility.Utility; /** @@ -79,7 +82,7 @@ import com.rabbitmq.utility.Utility; * @author Alex Panchenko * @author Johno Crawford */ -public class BlockingQueueConsumer { +public class BlockingQueueConsumer implements RecoveryListener { private static Log logger = LogFactory.getLog(BlockingQueueConsumer.class); @@ -150,6 +153,10 @@ public class BlockingQueueConsumer { private volatile boolean normalCancel; + volatile Thread thread; + + volatile boolean declaring; + /** * Create a consumer. The consumer must not attempt to use * the connection factory or communicate with the broker @@ -424,7 +431,8 @@ public class BlockingQueueConsumer { protected boolean cancelled() { return this.cancelled.get() || (this.abortStarted > 0 && - this.abortStarted + this.shutdownTimeout > System.currentTimeMillis()); + this.abortStarted + this.shutdownTimeout > System.currentTimeMillis()) + || !this.activeObjectCounter.isActive(); } /** @@ -559,10 +567,14 @@ public class BlockingQueueConsumer { if (logger.isDebugEnabled()) { logger.debug("Starting consumer " + this); } + + this.thread = Thread.currentThread(); + try { this.resourceHolder = ConnectionFactoryUtils.getTransactionalResourceHolder(this.connectionFactory, this.transactional); this.channel = this.resourceHolder.getChannel(); + addRecoveryListener(); } catch (AmqpAuthenticationException e) { throw new FatalListenerStartupException("Authentication failure", e); @@ -573,7 +585,11 @@ public class BlockingQueueConsumer { // mirrored queue might be being moved int passiveDeclareRetries = this.declarationRetries; + this.declaring = true; do { + if (cancelled()) { + break; + } try { attemptPassiveDeclarations(); if (passiveDeclareRetries < this.declarationRetries && logger.isInfoEnabled()) { @@ -589,7 +605,10 @@ public class BlockingQueueConsumer { Thread.sleep(this.failedDeclarationRetryInterval); } catch (InterruptedException e1) { + this.declaring = false; Thread.currentThread().interrupt(); + this.activeObjectCounter.release(this); + throw RabbitExceptionTranslator.convertRabbitAccessException(e1); } } } @@ -602,15 +621,17 @@ public class BlockingQueueConsumer { this.lastRetryDeclaration = System.currentTimeMillis(); } else { + this.declaring = false; this.activeObjectCounter.release(this); throw new QueuesNotAvailableException("Cannot prepare queue for listener. " + "Either the queue doesn't exist or the broker will not allow us to use it.", e); } } } - while (passiveDeclareRetries-- > 0); + while (passiveDeclareRetries-- > 0 && !cancelled()); + this.declaring = false; - if (!this.acknowledgeMode.isAutoAck()) { + if (!this.acknowledgeMode.isAutoAck() && !cancelled()) { // Set basicQos before calling basicConsume (otherwise if we are not acking the broker // will send blocks of 100 messages) try { @@ -624,9 +645,11 @@ public class BlockingQueueConsumer { try { - for (String queueName : this.queues) { - if (!this.missingQueues.contains(queueName)) { - consumeFromQueue(queueName); + if (!cancelled()) { + for (String queueName : this.queues) { + if (!this.missingQueues.contains(queueName)) { + consumeFromQueue(queueName); + } } } } @@ -635,6 +658,19 @@ public class BlockingQueueConsumer { } } + /** + * Add a listener if necessary so we can immediately close an autorecovered + * channel if necessary since the async consumer will no longer exist. + */ + private void addRecoveryListener() { + if (this.channel instanceof ChannelProxy) { + if (((ChannelProxy) this.channel).getTargetChannel() instanceof AutorecoveringChannel) { + ((AutorecoveringChannel) ((ChannelProxy) this.channel).getTargetChannel()) + .addRecoveryListener(this); + } + } + } + private void consumeFromQueue(String queue) throws IOException { String consumerTag = this.channel.basicConsume(queue, this.acknowledgeMode.isAutoAck(), (this.tagStrategy != null ? this.tagStrategy.createConsumerTag(queue) : ""), this.noLocal, this.exclusive, @@ -810,6 +846,28 @@ public class BlockingQueueConsumer { } + @Override + public void handleRecovery(Recoverable recoverable) { + // should never get here + handleRecoveryStarted(recoverable); + } + + @Override + public void handleRecoveryStarted(Recoverable recoverable) { + if (logger.isDebugEnabled()) { + logger.debug("Closing an autorecovered channel: " + recoverable); + } + try { + ((Channel) recoverable).close(); + } + catch (IOException e) { + logger.debug("Error closing an autorecovered channel"); + } + catch (TimeoutException e) { + logger.debug("Error closing an autorecovered channel"); + } + } + @Override public String toString() { return "Consumer@" + ObjectUtils.getIdentityHexString(this) + ": " 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 7905b90d..9a1344d9 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 @@ -27,11 +27,14 @@ import java.util.List; import java.util.Map; import java.util.Properties; import java.util.Set; +import java.util.concurrent.BlockingQueue; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; +import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; import org.aopalliance.aop.Advice; import org.apache.commons.logging.Log; @@ -127,6 +130,11 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta private volatile int prefetchCount = DEFAULT_PREFETCH_COUNT; + private final AtomicReference containerStoppingForAbort = new AtomicReference(); + + private final BlockingQueue abortEvents = + new LinkedBlockingQueue(); + private volatile long startConsumerMinInterval = DEFAULT_START_CONSUMER_MIN_INTERVAL; private volatile long stopConsumerMinInterval = DEFAULT_STOP_CONSUMER_MIN_INTERVAL; @@ -862,12 +870,13 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta checkMismatchedQueues(); super.doStart(); synchronized (this.consumersMonitor) { + if (this.consumers != null) { + throw new IllegalStateException("A stopped container should not have consumers"); + } int newConsumers = initializeConsumers(); if (this.consumers == null) { - if (logger.isInfoEnabled()) { - logger.info("Consumers were initialized and then cleared " + - "(presumably the container was stopped concurrently)"); - } + logger.info("Consumers were initialized and then cleared " + + "(presumably the container was stopped concurrently)"); return; } if (newConsumers <= 0) { @@ -902,8 +911,9 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta @Override protected void doShutdown() { - - if (!this.isRunning()) { + Thread thread = this.containerStoppingForAbort.get(); + if (thread != null && !thread.equals(Thread.currentThread())) { + logger.info("Shutdown ignored - container is stopping due to an aborted consumer"); return; } @@ -917,8 +927,15 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta consumer.basicCancel(true); canceledConsumers.add(consumer); consumerIterator.remove(); + if (consumer.declaring) { + consumer.thread.interrupt(); + } } } + else { + logger.info("Shutdown ignored - container is already stopped"); + return; + } } logger.info("Waiting for workers to finish."); boolean finished = this.cancellationLock.await(this.shutdownTimeout, TimeUnit.MILLISECONDS); @@ -941,6 +958,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta synchronized (this.consumersMonitor) { this.consumers = null; + this.cancellationLock.deactivate(); } } @@ -1136,6 +1154,10 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta // we haven't counted down yet) this.cancellationLock.release(consumer); this.consumers.remove(consumer); + if (!isActive()) { + // Do not restart - container is stopping + return; + } BlockingQueueConsumer newConsumer = createBlockingQueueConsumer(); newConsumer.setBackOffExecution(consumer.getBackOffExecution()); consumer = newConsumer; @@ -1398,7 +1420,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta * @throws InterruptedException if the consumer startup is interrupted */ private FatalListenerStartupException getStartupException() throws TimeoutException, - InterruptedException { + InterruptedException { if (!this.start.await( SimpleMessageListenerContainer.this.consumerStartTimeout, TimeUnit.MILLISECONDS)) { logger.error("Consumer failed to start in " @@ -1411,6 +1433,9 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta @Override public void run() { + if (!isActive()) { + return; + } boolean aborted = false; @@ -1609,7 +1634,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta catch (Error e) { //NOSONAR // ok to catch Error - we're aborting so will stop logger.error("Consumer thread error, thread abort.", e); - logConsumerException(e); + publishConsumerFailedEvent("Consumer threw an Error", true, e); aborted = true; } catch (Throwable t) { //NOSONAR @@ -1640,9 +1665,25 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta catch (AmqpException e) { logger.info("Could not cancel message consumer", e); } - if (aborted) { + if (aborted && SimpleMessageListenerContainer.this.containerStoppingForAbort + .compareAndSet(null, Thread.currentThread())) { logger.error("Stopping container from aborted consumer"); stop(); + SimpleMessageListenerContainer.this.containerStoppingForAbort.set(null); + ListenerContainerConsumerFailedEvent event = null; + do { + try { + event = SimpleMessageListenerContainer.this.abortEvents.poll(5, TimeUnit.SECONDS); + if (event != null) { + publishConsumerFailedEvent( + event.getReason(), event.isFatal(), event.getThrowable()); + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + while (event != null); } } else { @@ -1680,10 +1721,21 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta } private void publishConsumerFailedEvent(String reason, boolean fatal, Throwable t) { - if (SimpleMessageListenerContainer.this.applicationEventPublisher != null) { - SimpleMessageListenerContainer.this.applicationEventPublisher - .publishEvent(new ListenerContainerConsumerFailedEvent(SimpleMessageListenerContainer.this, - reason, t, fatal)); + if (!fatal || !isRunning()) { + if (SimpleMessageListenerContainer.this.applicationEventPublisher != null) { + SimpleMessageListenerContainer.this.applicationEventPublisher + .publishEvent(new ListenerContainerConsumerFailedEvent( + SimpleMessageListenerContainer.this, reason, t, fatal)); + } + } + else { + try { + SimpleMessageListenerContainer.this.abortEvents + .put(new ListenerContainerConsumerFailedEvent(this, reason, t, fatal)); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } } }