Skip to content

Commit 907d5cb

Browse files
garyrussellartembilan
authored andcommitted
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**
1 parent 8c43c57 commit 907d5cb

File tree

2 files changed

+82
-5
lines changed

2 files changed

+82
-5
lines changed

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

Lines changed: 47 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,9 @@
4848
import org.springframework.context.ApplicationEventPublisher;
4949
import org.springframework.context.ApplicationEventPublisherAware;
5050
import org.springframework.jmx.export.annotation.ManagedOperation;
51+
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
52+
import org.springframework.retry.policy.SimpleRetryPolicy;
53+
import org.springframework.retry.support.RetryTemplate;
5154
import org.springframework.util.Assert;
5255

5356
import com.rabbitmq.client.AMQP.Queue.DeclareOk;
@@ -96,13 +99,17 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat
9699

97100
private final RabbitTemplate rabbitTemplate;
98101

102+
private RetryTemplate retryTemplate;
103+
104+
private boolean retryDisabled;
105+
99106
private volatile boolean running = false;
100107

101-
private volatile boolean autoStartup = true;
108+
private boolean autoStartup = true;
102109

103-
private volatile ApplicationContext applicationContext;
110+
private ApplicationContext applicationContext;
104111

105-
private volatile boolean ignoreDeclarationExceptions;
112+
private boolean ignoreDeclarationExceptions;
106113

107114
private final Object lifecycleMonitor = new Object();
108115

@@ -367,6 +374,25 @@ public Properties getQueueProperties(final String queueName) {
367374
});
368375
}
369376

377+
/**
378+
* Set a retry template for auto declarations. There is a race condition with
379+
* auto-delete, exclusive queues in that the queue might still exist for a short time,
380+
* preventing the redeclaration. The default retry configuration will try 5 times with
381+
* an exponential backOff starting at 1 second a multiplier of 2.0 and a max interval
382+
* of 5 seconds. To disable retry, set the argument to {@code null}. Note that this
383+
* retry is at the macro level - all declarations will be retried within the scope of
384+
* this template. If you supplied a {@link RabbitTemplate} that is configured with a
385+
* {@link RetryTemplate}, its template will retry each individual declaration.
386+
* @param retryTemplate the retry template.
387+
* @since 1.7.8
388+
*/
389+
public void setRetryTemplate(RetryTemplate retryTemplate) {
390+
this.retryTemplate = retryTemplate;
391+
if (retryTemplate == null) {
392+
this.retryDisabled = true;
393+
}
394+
}
395+
370396
// Lifecycle implementation
371397

372398
public boolean isAutoStartup() {
@@ -391,6 +417,15 @@ public void afterPropertiesSet() {
391417
return;
392418
}
393419

420+
if (this.retryTemplate == null && !this.retryDisabled) {
421+
this.retryTemplate = new RetryTemplate();
422+
this.retryTemplate.setRetryPolicy(new SimpleRetryPolicy(5));
423+
ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
424+
backOffPolicy.setInitialInterval(1000);
425+
backOffPolicy.setMultiplier(2.0);
426+
backOffPolicy.setMaxInterval(5000);
427+
this.retryTemplate.setBackOffPolicy(backOffPolicy);
428+
}
394429
if (this.connectionFactory instanceof CachingConnectionFactory &&
395430
((CachingConnectionFactory) this.connectionFactory).getCacheMode() == CacheMode.CONNECTION) {
396431
this.logger.warn("RabbitAdmin auto declaration is not supported with CacheMode.CONNECTION");
@@ -413,7 +448,15 @@ public void afterPropertiesSet() {
413448
* chatter). In fact it might even be a good thing: exclusive queues only make sense if they are
414449
* declared for every connection. If anyone has a problem with it: use auto-startup="false".
415450
*/
416-
initialize();
451+
if (this.retryTemplate != null) {
452+
this.retryTemplate.execute(c -> {
453+
initialize();
454+
return null;
455+
});
456+
}
457+
else {
458+
initialize();
459+
}
417460
}
418461
finally {
419462
initializing.compareAndSet(true, false);

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

Lines changed: 35 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,9 @@
2828
import static org.junit.Assert.assertTrue;
2929
import static org.junit.Assert.fail;
3030
import static org.mockito.ArgumentMatchers.any;
31+
import static org.mockito.ArgumentMatchers.anyBoolean;
3132
import static org.mockito.ArgumentMatchers.anyString;
33+
import static org.mockito.ArgumentMatchers.isNull;
3234
import static org.mockito.BDDMockito.given;
3335
import static org.mockito.BDDMockito.willDoNothing;
3436
import static org.mockito.BDDMockito.willReturn;
@@ -52,11 +54,13 @@
5254
import java.util.concurrent.TimeoutException;
5355

5456
import org.apache.commons.logging.Log;
57+
import org.junit.Ignore;
5558
import org.junit.Rule;
5659
import org.junit.Test;
5760
import org.junit.rules.ExpectedException;
5861
import org.mockito.ArgumentCaptor;
5962

63+
import org.springframework.amqp.UncategorizedAmqpException;
6064
import org.springframework.amqp.core.AnonymousQueue;
6165
import org.springframework.amqp.core.Binding;
6266
import org.springframework.amqp.core.Binding.DestinationType;
@@ -284,9 +288,10 @@ public void testAvoidHangAMQP_508() {
284288
String longName = new String(new byte[300]).replace('\u0000', 'x');
285289
try {
286290
admin.declareQueue(new Queue(longName));
291+
fail("expected exception");
287292
}
288293
catch (Exception e) {
289-
e.printStackTrace();
294+
// NOSONAR
290295
}
291296
String goodName = "foobar";
292297
admin.declareQueue(new Queue(goodName));
@@ -352,6 +357,35 @@ public void testWithinInvoke() throws Exception {
352357
verifyZeroInteractions(channel2);
353358
}
354359

360+
@Test
361+
@Ignore // too long; not much value
362+
public void testRetry() throws Exception {
363+
com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
364+
com.rabbitmq.client.Connection connection = mock(com.rabbitmq.client.Connection.class);
365+
given(rabbitConnectionFactory.newConnection((ExecutorService) isNull(), anyString())).willReturn(connection);
366+
Channel channel = mock(Channel.class);
367+
given(connection.createChannel()).willReturn(channel);
368+
given(channel.isOpen()).willReturn(true);
369+
willThrow(new RuntimeException()).given(channel)
370+
.queueDeclare(anyString(), anyBoolean(), anyBoolean(), anyBoolean(), isNull());
371+
CachingConnectionFactory ccf = new CachingConnectionFactory(rabbitConnectionFactory);
372+
RabbitAdmin admin = new RabbitAdmin(ccf);
373+
GenericApplicationContext ctx = new GenericApplicationContext();
374+
ctx.getBeanFactory().registerSingleton("foo", new AnonymousQueue());
375+
ctx.getBeanFactory().registerSingleton("admin", admin);
376+
admin.setApplicationContext(ctx);
377+
ctx.getBeanFactory().initializeBean(admin, "admin");
378+
ctx.refresh();
379+
try {
380+
ccf.createConnection();
381+
fail("expected exception");
382+
}
383+
catch (UncategorizedAmqpException e) {
384+
// NOSONAR
385+
}
386+
ctx.close();
387+
}
388+
355389
@Configuration
356390
public static class Config {
357391

0 commit comments

Comments
 (0)