From 1ced9ab0610a7bc3b9521e13e7ada89a3734a82e Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 22 May 2018 15:20:43 -0400 Subject: [PATCH] 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 --- .../amqp/rabbit/core/RabbitAdmin.java | 77 +++++++++++++++++-- .../amqp/rabbit/core/RabbitAdminTests.java | 38 ++++++++- 2 files changed, 106 insertions(+), 9 deletions(-) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java index e30b2e6e..a098914f 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java @@ -49,6 +49,11 @@ import org.springframework.context.ApplicationContextAware; 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 class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat public void declareExchange(final Exchange exchange) { try { this.rabbitTemplate.execute(new ChannelCallback() { + @Override public Object doInRabbit(Channel channel) throws Exception { declareExchanges(channel, exchange); @@ -185,6 +195,7 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat @ManagedOperation public boolean deleteExchange(final String exchangeName) { return this.rabbitTemplate.execute(new ChannelCallback() { + @Override public Boolean doInRabbit(Channel channel) throws Exception { if (isDeletingDefaultExchange(exchangeName)) { @@ -219,6 +230,7 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat public String declareQueue(final Queue queue) { try { return this.rabbitTemplate.execute(new ChannelCallback() { + @Override public String doInRabbit(Channel channel) throws Exception { DeclareOk[] declared = declareQueues(channel, queue); @@ -244,6 +256,7 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat public Queue declareQueue() { try { DeclareOk declareOk = this.rabbitTemplate.execute(new ChannelCallback() { + @Override public DeclareOk doInRabbit(Channel channel) throws Exception { return channel.queueDeclare(); @@ -262,6 +275,7 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat @ManagedOperation public boolean deleteQueue(final String queueName) { return this.rabbitTemplate.execute(new ChannelCallback() { + @Override public Boolean doInRabbit(Channel channel) throws Exception { try { @@ -279,6 +293,7 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat @ManagedOperation public void deleteQueue(final String queueName, final boolean unused, final boolean empty) { this.rabbitTemplate.execute(new ChannelCallback() { + @Override public Object doInRabbit(Channel channel) throws Exception { channel.queueDelete(queueName, unused, empty); @@ -291,6 +306,7 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat @ManagedOperation public void purgeQueue(final String queueName, final boolean noWait) { this.rabbitTemplate.execute(new ChannelCallback() { + @Override public Object doInRabbit(Channel channel) throws Exception { channel.queuePurge(queueName); @@ -305,6 +321,7 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat public void declareBinding(final Binding binding) { try { this.rabbitTemplate.execute(new ChannelCallback() { + @Override public Object doInRabbit(Channel channel) throws Exception { declareBindings(channel, binding); @@ -321,6 +338,7 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat @ManagedOperation public void removeBinding(final Binding binding) { this.rabbitTemplate.execute(new ChannelCallback() { + @Override public Object doInRabbit(Channel channel) throws Exception { if (binding.isDestinationQueue()) { @@ -348,6 +366,7 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat public Properties getQueueProperties(final String queueName) { Assert.hasText(queueName, "'queueName' cannot be null or empty"); return this.rabbitTemplate.execute(new ChannelCallback() { + @Override public Properties doInRabbit(Channel channel) throws Exception { try { @@ -382,6 +401,25 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat }); } + /** + * 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 class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat 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 class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat * 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() { + + @Override + public Object doWithRetry(RetryContext c) throws RuntimeException { + initialize(); + return null; + } + }); + } + else { + initialize(); + } } finally { initializing.compareAndSet(true, false); @@ -469,8 +529,8 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat @SuppressWarnings("rawtypes") Collection collections = this.declareCollections - ? this.applicationContext.getBeansOfType(Collection.class, false, false).values() - : Collections.emptyList(); + ? this.applicationContext.getBeansOfType(Collection.class, false, false).values() + : Collections.emptyList(); for (Collection collection : collections) { if (collection.size() > 0 && collection.iterator().next() instanceof Declarable) { for (Object declarable : collection) { @@ -492,7 +552,7 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat final Collection 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 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat return; } this.rabbitTemplate.execute(new ChannelCallback() { + @Override public Object doInRabbit(Channel channel) throws Exception { declareExchanges(channel, exchanges.toArray(new Exchange[exchanges.size()])); @@ -540,7 +601,7 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat for (T declarable : declarables) { Collection adminsWithWhichToDeclare = declarable.getDeclaringAdmins(); if (declarable.shouldDeclare() && - (adminsWithWhichToDeclare.isEmpty() || adminsWithWhichToDeclare.contains(this))) { + (adminsWithWhichToDeclare.isEmpty() || adminsWithWhichToDeclare.contains(this))) { filtered.add(declarable); } } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java index 2b94f11f..6813c7f9 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java @@ -26,8 +26,12 @@ import static org.junit.Assert.assertSame; 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 java.util.concurrent.TimeoutException; 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 class RabbitAdminTests { 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 class RabbitAdminTests { 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 {