AMQP-814: Add retry to RabbitAdmin

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

Add retry to avoid race conditions with auto-delete, exclusive queues.

**cherry-pick to 2.0.x**
**back port to 1.7.x, without lambda in RabbitAdmin**

# Conflicts:
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java
#	spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java

Author: garyrussell

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java

@@ -49,6 +49,11 @@
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.jmx.export.annotation.ManagedOperation;
+import org.springframework.retry.RetryCallback;
+import org.springframework.retry.RetryContext;
+import org.springframework.retry.backoff.ExponentialBackOffPolicy;
+import org.springframework.retry.policy.SimpleRetryPolicy;
+import org.springframework.retry.support.RetryTemplate;
 import org.springframework.util.Assert;
 
 import com.rabbitmq.client.AMQP.Queue.DeclareOk;
@@ -97,13 +102,17 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat
 
 	private final RabbitTemplate rabbitTemplate;
 
+	private RetryTemplate retryTemplate;
+
+	private boolean retryDisabled;
+
 	private volatile boolean running = false;
 
-	private volatile boolean autoStartup = true;
+	private boolean autoStartup = true;
 
-	private volatile ApplicationContext applicationContext;
+	private ApplicationContext applicationContext;
 
-	private volatile boolean ignoreDeclarationExceptions;
+	private boolean ignoreDeclarationExceptions;
 
 	private final Object lifecycleMonitor = new Object();
 
@@ -169,6 +178,7 @@ public RabbitTemplate getRabbitTemplate() {
 	public void declareExchange(final Exchange exchange) {
 		try {
 			this.rabbitTemplate.execute(new ChannelCallback<Object>() {
+
 				@Override
 				public Object doInRabbit(Channel channel) throws Exception {
 					declareExchanges(channel, exchange);
@@ -185,6 +195,7 @@ public Object doInRabbit(Channel channel) throws Exception {
 	@ManagedOperation
 	public boolean deleteExchange(final String exchangeName) {
 		return this.rabbitTemplate.execute(new ChannelCallback<Boolean>() {
+
 			@Override
 			public Boolean doInRabbit(Channel channel) throws Exception {
 				if (isDeletingDefaultExchange(exchangeName)) {
@@ -219,6 +230,7 @@ public Boolean doInRabbit(Channel channel) throws Exception {
 	public String declareQueue(final Queue queue) {
 		try {
 			return this.rabbitTemplate.execute(new ChannelCallback<String>() {
+
 				@Override
 				public String doInRabbit(Channel channel) throws Exception {
 					DeclareOk[] declared = declareQueues(channel, queue);
@@ -244,6 +256,7 @@ public String doInRabbit(Channel channel) throws Exception {
 	public Queue declareQueue() {
 		try {
 			DeclareOk declareOk = this.rabbitTemplate.execute(new ChannelCallback<DeclareOk>() {
+
 				@Override
 				public DeclareOk doInRabbit(Channel channel) throws Exception {
 					return channel.queueDeclare();
@@ -262,6 +275,7 @@ public DeclareOk doInRabbit(Channel channel) throws Exception {
 	@ManagedOperation
 	public boolean deleteQueue(final String queueName) {
 		return this.rabbitTemplate.execute(new ChannelCallback<Boolean>() {
+
 			@Override
 			public Boolean doInRabbit(Channel channel) throws Exception {
 				try {
@@ -279,6 +293,7 @@ public Boolean doInRabbit(Channel channel) throws Exception {
 	@ManagedOperation
 	public void deleteQueue(final String queueName, final boolean unused, final boolean empty) {
 		this.rabbitTemplate.execute(new ChannelCallback<Object>() {
+
 			@Override
 			public Object doInRabbit(Channel channel) throws Exception {
 				channel.queueDelete(queueName, unused, empty);
@@ -291,6 +306,7 @@ public Object doInRabbit(Channel channel) throws Exception {
 	@ManagedOperation
 	public void purgeQueue(final String queueName, final boolean noWait) {
 		this.rabbitTemplate.execute(new ChannelCallback<Object>() {
+
 			@Override
 			public Object doInRabbit(Channel channel) throws Exception {
 				channel.queuePurge(queueName);
@@ -305,6 +321,7 @@ public Object doInRabbit(Channel channel) throws Exception {
 	public void declareBinding(final Binding binding) {
 		try {
 			this.rabbitTemplate.execute(new ChannelCallback<Object>() {
+
 				@Override
 				public Object doInRabbit(Channel channel) throws Exception {
 					declareBindings(channel, binding);
@@ -321,6 +338,7 @@ public Object doInRabbit(Channel channel) throws Exception {
 	@ManagedOperation
 	public void removeBinding(final Binding binding) {
 		this.rabbitTemplate.execute(new ChannelCallback<Object>() {
+
 			@Override
 			public Object doInRabbit(Channel channel) throws Exception {
 				if (binding.isDestinationQueue()) {
@@ -348,6 +366,7 @@ public Object doInRabbit(Channel channel) throws Exception {
 	public Properties getQueueProperties(final String queueName) {
 		Assert.hasText(queueName, "'queueName' cannot be null or empty");
 		return this.rabbitTemplate.execute(new ChannelCallback<Properties>() {
+
 			@Override
 			public Properties doInRabbit(Channel channel) throws Exception {
 				try {
@@ -382,6 +401,25 @@ public Properties doInRabbit(Channel channel) throws Exception {
 		});
 	}
 
+	/**
+	 * Set a retry template for auto declarations. There is a race condition with
+	 * auto-delete, exclusive queues in that the queue might still exist for a short time,
+	 * preventing the redeclaration. The default retry configuration will try 5 times with
+	 * an exponential backOff starting at 1 second a multiplier of 2.0 and a max interval
+	 * of 5 seconds. To disable retry, set the argument to {@code null}. Note that this
+	 * retry is at the macro level - all declarations will be retried within the scope of
+	 * this template. If you supplied a {@link RabbitTemplate} that is configured with a
+	 * {@link RetryTemplate}, its template will retry each individual declaration.
+	 * @param retryTemplate the retry template.
+	 * @since 1.7.8
+	 */
+	public void setRetryTemplate(RetryTemplate retryTemplate) {
+		this.retryTemplate = retryTemplate;
+		if (retryTemplate == null) {
+			this.retryDisabled = true;
+		}
+	}
+
 	// Lifecycle implementation
 
 	public boolean isAutoStartup() {
@@ -406,6 +444,15 @@ public void afterPropertiesSet() {
 				return;
 			}
 
+			if (this.retryTemplate == null && !this.retryDisabled) {
+				this.retryTemplate = new RetryTemplate();
+				this.retryTemplate.setRetryPolicy(new SimpleRetryPolicy(5));
+				ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
+				backOffPolicy.setInitialInterval(1000);
+				backOffPolicy.setMultiplier(2.0);
+				backOffPolicy.setMaxInterval(5000);
+				this.retryTemplate.setBackOffPolicy(backOffPolicy);
+			}
 			if (this.connectionFactory instanceof CachingConnectionFactory &&
 					((CachingConnectionFactory) this.connectionFactory).getCacheMode() == CacheMode.CONNECTION) {
 				this.logger.warn("RabbitAdmin auto declaration is not supported with CacheMode.CONNECTION");
@@ -430,7 +477,20 @@ public void onCreate(Connection connection) {
 						 * chatter). In fact it might even be a good thing: exclusive queues only make sense if they are
 						 * declared for every connection. If anyone has a problem with it: use auto-startup="false".
 						 */
-						initialize();
+						if (RabbitAdmin.this.retryTemplate != null) {
+							RabbitAdmin.this.retryTemplate.execute(
+									new RetryCallback<Object, RuntimeException>() {
+
+										@Override
+										public Object doWithRetry(RetryContext c) throws RuntimeException {
+											initialize();
+											return null;
+										}
+									});
+						}
+						else {
+							initialize();
+						}
 					}
 					finally {
 						initializing.compareAndSet(true, false);
@@ -469,8 +529,8 @@ public void initialize() {
 
 		@SuppressWarnings("rawtypes")
 		Collection<Collection> collections = this.declareCollections
-			? this.applicationContext.getBeansOfType(Collection.class, false, false).values()
-			: Collections.<Collection>emptyList();
+				? this.applicationContext.getBeansOfType(Collection.class, false, false).values()
+				: Collections.<Collection>emptyList();
 		for (Collection<?> collection : collections) {
 			if (collection.size() > 0 && collection.iterator().next() instanceof Declarable) {
 				for (Object declarable : collection) {
@@ -492,7 +552,7 @@ else if (declarable instanceof Binding) {
 		final Collection<Binding> bindings = filterDeclarables(contextBindings);
 
 		for (Exchange exchange : exchanges) {
-			if ((!exchange.isDurable() || exchange.isAutoDelete())  && this.logger.isInfoEnabled()) {
+			if ((!exchange.isDurable() || exchange.isAutoDelete()) && this.logger.isInfoEnabled()) {
 				this.logger.info("Auto-declaring a non-durable or auto-delete Exchange ("
 						+ exchange.getName()
 						+ ") durable:" + exchange.isDurable() + ", auto-delete:" + exchange.isAutoDelete() + ". "
@@ -517,6 +577,7 @@ else if (declarable instanceof Binding) {
 			return;
 		}
 		this.rabbitTemplate.execute(new ChannelCallback<Object>() {
+
 			@Override
 			public Object doInRabbit(Channel channel) throws Exception {
 				declareExchanges(channel, exchanges.toArray(new Exchange[exchanges.size()]));
@@ -540,7 +601,7 @@ private <T extends Declarable> Collection<T> filterDeclarables(Collection<T> dec
 		for (T declarable : declarables) {
 			Collection<?> adminsWithWhichToDeclare = declarable.getDeclaringAdmins();
 			if (declarable.shouldDeclare() &&
-				(adminsWithWhichToDeclare.isEmpty() || adminsWithWhichToDeclare.contains(this))) {
+					(adminsWithWhichToDeclare.isEmpty() || adminsWithWhichToDeclare.contains(this))) {
 				filtered.add(declarable);
 			}
 		}

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java

@@ -26,8 +26,12 @@
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.BDDMockito.willThrow;
 import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyBoolean;
 import static org.mockito.Matchers.anyString;
+import static org.mockito.Matchers.isNull;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.doThrow;
@@ -50,12 +54,14 @@
 
 import org.apache.commons.logging.Log;
 import org.hamcrest.Matchers;
+import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
 import org.mockito.ArgumentCaptor;
 import org.mockito.internal.stubbing.answers.DoesNothing;
 
+import org.springframework.amqp.UncategorizedAmqpException;
 import org.springframework.amqp.core.AnonymousQueue;
 import org.springframework.amqp.core.Binding;
 import org.springframework.amqp.core.Binding.DestinationType;
@@ -270,9 +276,10 @@ public void testAvoidHangAMQP_508() {
 		String longName = new String(new byte[300]).replace('\u0000', 'x');
 		try {
 			admin.declareQueue(new Queue(longName));
+			fail("expected exception");
 		}
 		catch (Exception e) {
-			e.printStackTrace();
+			// NOSONAR
 		}
 		String goodName = "foobar";
 		admin.declareQueue(new Queue(goodName));
@@ -311,6 +318,35 @@ public void testIgnoreDeclarationExeptionsTimeout() throws Exception {
 		assertSame(events.get(3), admin.getLastDeclarationExceptionEvent());
 	}
 
+	@Test
+	@Ignore // too long; not much value
+	public void testRetry() throws Exception {
+		com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
+		com.rabbitmq.client.Connection connection = mock(com.rabbitmq.client.Connection.class);
+		given(rabbitConnectionFactory.newConnection((ExecutorService) isNull(), anyString())).willReturn(connection);
+		Channel channel = mock(Channel.class);
+		given(connection.createChannel()).willReturn(channel);
+		given(channel.isOpen()).willReturn(true);
+		willThrow(new RuntimeException()).given(channel)
+				.queueDeclare(anyString(), anyBoolean(), anyBoolean(), anyBoolean(), any());
+		CachingConnectionFactory ccf = new CachingConnectionFactory(rabbitConnectionFactory);
+		RabbitAdmin admin = new RabbitAdmin(ccf);
+		GenericApplicationContext ctx = new GenericApplicationContext();
+		ctx.getBeanFactory().registerSingleton("foo", new AnonymousQueue());
+		ctx.getBeanFactory().registerSingleton("admin", admin);
+		admin.setApplicationContext(ctx);
+		ctx.getBeanFactory().initializeBean(admin, "admin");
+		ctx.refresh();
+		try {
+			ccf.createConnection();
+			fail("expected exception");
+		}
+		catch (UncategorizedAmqpException e) {
+			// NOSONAR
+		}
+		ctx.close();
+	}
+
 	@Configuration
 	public static class Config {