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

@@ -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 {

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) {

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 <T> the object type.
+ *
  * @author Dave Syer
+ * @author Artem Bilan
  *
  */
 public class ActiveObjectCounter<T> {
 
 	private final ConcurrentMap<T, CountDownLatch> locks = new ConcurrentHashMap<T, CountDownLatch>();
 
+	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<T> {
 
 	public void reset() {
 		this.locks.clear();
+		this.active = true;
+	}
+
+	public void deactivate() {
+		this.active = false;
+	}
+
+	public boolean isActive() {
+		return this.active;
 	}
 
 }

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) + ": "

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<Thread> containerStoppingForAbort = new AtomicReference<Thread>();
+
+	private final BlockingQueue<ListenerContainerConsumerFailedEvent> abortEvents =
+			new LinkedBlockingQueue<ListenerContainerConsumerFailedEvent>();
+
 	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();
+				}
 			}
 		}