AMQP-472: Recover From Conn. Close During Init

JIRA: https://jira.spring.io/browse/AMQP-472

Previously, any `IOException` during passive queue declaration would enter
declaration retry and eventually throw a `QueuesNotAvailableException`. Whether
or not that is recoverable depends on the container's `missingQueuesFatal` property.

If the `IOException` is due to a connection close, we should not try to redeclare
and, further, recovery should be unconditional.

When a queue declaration fails, check if the connection is open and, if not, throw
an `AmqpIOException`, causing container recovery to begin. If the connection is open
continue retrying queue declaration as before.

In addition, expose the queue declaration retry properties on the `SimpleMessageListenerContainer`.

AMQP-472: Polishing

Polishing - PR Comments

AMQP-472: Polishing to the last changes

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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. You may obtain a copy of the License at
@@ -198,7 +198,7 @@ public abstract class RabbitUtils {
 					&& ((AMQP.Channel.Close) shutdownReason).getMethodId() == 10); // declare
 	}
 
-	protected static Object determineShutdownReason(ShutdownSignalException sig) {
+	public static Object determineShutdownReason(ShutdownSignalException sig) {
 		if (shutDownSignalReasonMethod == null) {
 			return false;
 		}

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

@@ -113,7 +113,11 @@ public class BlockingQueueConsumer {
 
 	private final Set<String> missingQueues = Collections.synchronizedSet(new HashSet<String>());
 
-	private final long retryDeclarationInterval = 60000;
+	private long retryDeclarationInterval = 60000;
+
+	private long failedDeclarationRetryInterval = 5000;
+
+	private int declarationRetries = 3;
 
 	private long lastRetryDeclaration;
 
@@ -233,6 +237,36 @@ public class BlockingQueueConsumer {
 	public final void setQuiesce(long shutdownTimeout) {
 	}
 
+	/**
+	 * Set the number of retries after passive queue declaration fails.
+	 * @param declarationRetries The number of retries, default 3.
+	 * @see #setFailedDeclarationRetryInterval(long)
+	 * @since 1.3.9
+	 */
+	public void setDeclarationRetries(int declarationRetries) {
+		this.declarationRetries = declarationRetries;
+	}
+
+	/**
+	 * Set the interval between passive queue declaration attempts in milliseconds.
+	 * @param failedDeclarationRetryInterval the interval, default 5000.
+	 * @see #setDeclarationRetries(int)
+	 * @since 1.3.9
+	 */
+	public void setFailedDeclarationRetryInterval(long failedDeclarationRetryInterval) {
+		this.failedDeclarationRetryInterval = failedDeclarationRetryInterval;
+	}
+
+	/**
+	 * When consuming multiple queues, set the interval between declaration attempts when only
+	 * a subset of the queues were available (milliseconds).
+	 * @param retryDeclarationInterval the interval, default 60000.
+	 * @since 1.3.9
+	 */
+	public void setRetryDeclarationInterval(long retryDeclarationInterval) {
+		this.retryDeclarationInterval = retryDeclarationInterval;
+	}
+
 	protected void basicCancel() {
 		for (String consumerTag : this.consumerTags.keySet()) {
 			try {
@@ -396,21 +430,21 @@ public class BlockingQueueConsumer {
 		this.activeObjectCounter.add(this);
 
 		// mirrored queue might be being moved
-		int passiveDeclareTries = 3;
+		int passiveDeclareRetries = this.declarationRetries;
 		do {
 			try {
 				attemptPassiveDeclarations();
-				if (passiveDeclareTries < 3 && logger.isInfoEnabled()) {
+				if (passiveDeclareRetries < this.declarationRetries && logger.isInfoEnabled()) {
 					logger.info("Queue declaration succeeded after retrying");
 				}
-				passiveDeclareTries = 0;
+				passiveDeclareRetries = 0;
 			}
 			catch (DeclarationException e) {
-				if (passiveDeclareTries > 0 && channel.isOpen()) {
+				if (passiveDeclareRetries > 0 && channel.isOpen()) {
 					if (logger.isWarnEnabled()) {
-						logger.warn("Queue declaration failed; retries left=" + (passiveDeclareTries-1), e);
+						logger.warn("Queue declaration failed; retries left=" + (passiveDeclareRetries), e);
 						try {
-							Thread.sleep(5000);
+							Thread.sleep(this.failedDeclarationRetryInterval);
 						}
 						catch (InterruptedException e1) {
 							Thread.currentThread().interrupt();
@@ -432,7 +466,7 @@ public class BlockingQueueConsumer {
 				}
 			}
 		}
-		while (passiveDeclareTries-- > 0);
+		while (passiveDeclareRetries-- > 0);
 
 		if (!acknowledgeMode.isAutoAck()) {
 			// Set basicQos before calling basicConsume (otherwise if we are not acking the broker
@@ -483,8 +517,11 @@ public class BlockingQueueConsumer {
 				if (logger.isWarnEnabled()) {
 					logger.warn("Failed to declare queue:" + queueName);
 				}
+				if (!this.channel.isOpen()) {
+					throw new AmqpIOException(e);
+				}
 				if (failures == null) {
-					failures = new DeclarationException();
+					failures = new DeclarationException(e);
 				}
 				failures.addFailedQueue(queueName);
 			}
@@ -626,6 +663,10 @@ public class BlockingQueueConsumer {
 			super("Failed to declare queue(s):");
 		}
 
+		public DeclarationException(Throwable t) {
+			super("Failed to declare queue(s):", t);
+		}
+
 		private final List<String> failedQueues = new ArrayList<String>();
 
 		void addFailedQueue(String queue) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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. You may obtain a copy of the License at
@@ -170,6 +170,12 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 
 	private ContainerDelegate proxy = delegate;
 
+	private Integer declarationRetries;
+
+	private Long failedDeclarationRetryInterval;
+
+	private Long retryDeclarationInterval;
+
 	/**
 	 * Default constructor for convenient dependency injection via setters.
 	 */
@@ -546,6 +552,36 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 		}
 	}
 
+	/**
+	 * Set the number of retries after passive queue declaration fails.
+	 * @param declarationRetries The number of retries, default 3.
+	 * @see #setFailedDeclarationRetryInterval(long)
+	 * @since 1.3.9
+	 */
+	public void setDeclarationRetries(int declarationRetries) {
+		this.declarationRetries = declarationRetries;
+	}
+
+	/**
+	 * Set the interval between passive queue declaration attempts in milliseconds.
+	 * @param failedDeclarationRetryInterval the interval, default 5000.
+	 * @see #setDeclarationRetries(int)
+	 * @since 1.3.9
+	 */
+	public void setFailedDeclarationRetryInterval(long failedDeclarationRetryInterval) {
+		this.failedDeclarationRetryInterval = failedDeclarationRetryInterval;
+	}
+
+	/**
+	 * When consuming multiple queues, set the interval between declaration attempts when only
+	 * a subset of the queues were available (milliseconds).
+	 * @param retryDeclarationInterval the interval, default 60000.
+	 * @since 1.3.9
+	 */
+	public void setRetryDeclarationInterval(long retryDeclarationInterval) {
+		this.retryDeclarationInterval = retryDeclarationInterval;
+	}
+
 	/**
 	 * Avoid the possibility of not configuring the CachingConnectionFactory in sync with the number of concurrent
 	 * consumers.
@@ -852,6 +888,15 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 		consumer = new BlockingQueueConsumer(getConnectionFactory(), this.messagePropertiesConverter, cancellationLock,
 				getAcknowledgeMode(), isChannelTransacted(), actualPrefetchCount, this.defaultRequeueRejected,
 				this.consumerArgs, this.exclusive, queues);
+		if (this.declarationRetries != null) {
+			consumer.setDeclarationRetries(this.declarationRetries);
+		}
+		if (this.failedDeclarationRetryInterval != null) {
+			consumer.setFailedDeclarationRetryInterval(this.failedDeclarationRetryInterval);
+		}
+		if (this.retryDeclarationInterval != null) {
+			consumer.setRetryDeclarationInterval(this.retryDeclarationInterval);
+		}
 		return consumer;
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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.
@@ -113,6 +113,7 @@ public class BlockingQueueConsumerTests {
 
 		when(connectionFactory.createConnection()).thenReturn(connection);
 		when(connection.createChannel(Mockito.anyBoolean())).thenReturn(channel);
+		when(channel.isOpen()).thenReturn(true);
 		when(channel.queueDeclarePassive(Mockito.anyString()))
 				.then(new Answer<Object>() {
 
@@ -134,6 +135,9 @@ public class BlockingQueueConsumerTests {
 				new DefaultMessagePropertiesConverter(), new ActiveObjectCounter<BlockingQueueConsumer>(),
 				AcknowledgeMode.AUTO, true, 20, "good", "bad");
 
+		blockingQueueConsumer.setDeclarationRetries(1);
+		blockingQueueConsumer.setRetryDeclarationInterval(10);
+		blockingQueueConsumer.setFailedDeclarationRetryInterval(10);
 		blockingQueueConsumer.start();
 
 		verify(channel).basicQos(20);

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

@@ -1,29 +0,0 @@
-/*
- * Copyright 2002-2014 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. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
- * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations under the License.
- */
-package org.springframework.amqp.rabbit.listener;
-
-import org.springframework.amqp.rabbit.connection.ConnectionFactory;
-import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
-import org.springframework.amqp.rabbit.test.BrokerTestUtils;
-
-public class MessageListenerRecoverySingleConnectionIntegrationTests extends MessageListenerRecoveryCachingConnectionIntegrationTests {
-
-	@Override
-	protected ConnectionFactory createConnectionFactory() {
-		SingleConnectionFactory connectionFactory = new SingleConnectionFactory();
-		connectionFactory.setHost("localhost");
-		connectionFactory.setPort(BrokerTestUtils.getPort());
-		return connectionFactory;
-	}
-
-}

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

@@ -58,6 +58,7 @@ import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.context.support.GenericApplicationContext;
 
+import com.rabbitmq.client.AMQP.Queue.DeclareOk;
 import com.rabbitmq.client.Channel;
 
 /**
@@ -318,6 +319,86 @@ public class SimpleMessageListenerContainerIntegration2Tests {
 
 		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		assertTrue(networkGlitch.get());
+
+		container.stop();
+	}
+
+	@Test
+	public void testRestartConsumerOnConnectionLossDuringQueueDeclare() throws Exception {
+		this.template.convertAndSend(queue.getName(), "foo");
+
+		ConnectionFactory connectionFactory = new CachingConnectionFactory("localhost", BrokerTestUtils.getPort());
+
+		final AtomicBoolean networkGlitch = new AtomicBoolean();
+
+		class MockChannel extends PublisherCallbackChannelImpl {
+
+			public MockChannel(Channel delegate) {
+				super(delegate);
+			}
+
+			@Override
+			public DeclareOk queueDeclarePassive(String queue) throws IOException {
+				if (networkGlitch.compareAndSet(false, true)) {
+					getConnection().close();
+					throw new IOException("Intentional connection reset");
+				}
+				return super.queueDeclarePassive(queue);
+			}
+
+		}
+
+		Connection connection = spy(connectionFactory.createConnection());
+		when(connection.createChannel(anyBoolean())).then(new Answer<Channel>() {
+
+			@Override
+			public Channel answer(InvocationOnMock invocation) throws Throwable {
+				return new MockChannel((Channel) invocation.callRealMethod());
+			}
+
+		});
+
+		DirectFieldAccessor dfa = new DirectFieldAccessor(connectionFactory);
+		dfa.setPropertyValue("connection", connection);
+
+		CountDownLatch latch = new CountDownLatch(1);
+		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
+		container.setMessageListener(new MessageListenerAdapter(new PojoListener(latch)));
+		container.setQueueNames(queue.getName());
+		container.setRecoveryInterval(500);
+		container.afterPropertiesSet();
+		container.start();
+
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
+		assertTrue(networkGlitch.get());
+
+		container.stop();
+	}
+
+	@Test
+	public void testRestartConsumerMissingQueue() throws Exception {
+		Queue queue = new AnonymousQueue();
+		this.template.convertAndSend(queue.getName(), "foo");
+
+		ConnectionFactory connectionFactory = new CachingConnectionFactory("localhost", BrokerTestUtils.getPort());
+
+		CountDownLatch latch = new CountDownLatch(1);
+		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
+		container.setMessageListener(new MessageListenerAdapter(new PojoListener(latch)));
+		container.setQueues(queue);
+		container.setRecoveryInterval(500);
+		container.setMissingQueuesFatal(false);
+		container.setDeclarationRetries(1);
+		container.setFailedDeclarationRetryInterval(100);
+		container.afterPropertiesSet();
+		container.start();
+
+		new RabbitAdmin(connectionFactory).declareQueue(queue);
+		this.template.convertAndSend(queue.getName(), "foo");
+
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
+
+		container.stop();
 	}
 
 	private boolean containerStoppedForAbortWithBadListener() throws InterruptedException {