diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java index 618cb877c..45580a287 100644 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java @@ -55,6 +55,12 @@ public class RabbitConsumerProperties extends RabbitCommonProperties { */ private boolean exclusive; + private boolean missingQueuesFatal = false; + + private Integer queueDeclarationRetries; + + private Long failedDeclarationRetryInterval; + public boolean isTransacted() { return transacted; } @@ -173,4 +179,28 @@ public class RabbitConsumerProperties extends RabbitCommonProperties { this.exclusive = exclusive; } + public boolean getMissingQueuesFatal() { + return this.missingQueuesFatal; + } + + public void setMissingQueuesFatal(boolean missingQueuesFatal) { + this.missingQueuesFatal = missingQueuesFatal; + } + + public Integer getQueueDeclarationRetries() { + return this.queueDeclarationRetries; + } + + public void setQueueDeclarationRetries(Integer queueDeclarationRetries) { + this.queueDeclarationRetries = queueDeclarationRetries; + } + + public Long getFailedDeclarationRetryInterval() { + return this.failedDeclarationRetryInterval; + } + + public void setFailedDeclarationRetryInterval(Long failedDeclarationRetryInterval) { + this.failedDeclarationRetryInterval = failedDeclarationRetryInterval; + } + } diff --git a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc index 6920094e9..701da4a54 100644 --- a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc +++ b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc @@ -198,6 +198,10 @@ expires:: how long before an unused queue is deleted (ms) + Default: `no expiration` +failedDeclarationRetryInterval:: + The interval (ms) between attempts to consume from a queue if it is missing. ++ +Default: 5000 headerPatterns:: Patterns for headers to be mapped from inbound messages. + @@ -224,6 +228,11 @@ maxPriority:: maximum priority of messages in the queue (0-255) + Default:: `none` +missingQueuesFatal:: + If the queue cannot be found, treat the condition as fatal and stop the listener container. + Defaults to `false` so that the container keeps trying to consume from the queue, for example when using a cluster and the node hosting a non HA queue is down. ++ +Default:: `false` prefetch:: Prefetch count. + @@ -232,6 +241,11 @@ prefix:: A prefix to be added to the name of the `destination` and queues. + Default: "". +queueDeclarationRetries:: + The number of times to retry consuming from a queue if it is missing. + Only relevant if `missingQueuesFatal` is `true`; otherwise the container keeps retrying indefinitely. ++ +Default:: `3` queueNameGroupOnly:: When true, consume from a queue with a name equal to the `group`; otherwise the queue name is `destination.group`. This is useful, for example, when using Spring Cloud Stream to consume from an existing RabbitMQ queue. diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index 27907c772..e499d2164 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -394,8 +394,17 @@ public class RabbitMessageChannelBinder listenerContainer.setTaskExecutor(new SimpleAsyncTaskExecutor(consumerDestination.getName() + "-")); listenerContainer.setQueueNames(consumerDestination.getName()); listenerContainer.setAfterReceivePostProcessors(this.decompressingPostProcessor); - listenerContainer.setMessagePropertiesConverter(RabbitMessageChannelBinder.inboundMessagePropertiesConverter); + listenerContainer.setMessagePropertiesConverter( + RabbitMessageChannelBinder.inboundMessagePropertiesConverter); listenerContainer.setExclusive(properties.getExtension().isExclusive()); + listenerContainer.setMissingQueuesFatal(properties.getExtension().getMissingQueuesFatal()); + if (properties.getExtension().getQueueDeclarationRetries() != null) { + listenerContainer.setDeclarationRetries(properties.getExtension().getQueueDeclarationRetries()); + } + if (properties.getExtension().getFailedDeclarationRetryInterval() != null) { + listenerContainer.setFailedDeclarationRetryInterval( + properties.getExtension().getFailedDeclarationRetryInterval()); + } listenerContainer.afterPropertiesSet(); AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(listenerContainer); diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index c309d07e6..067a183fa 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java @@ -254,6 +254,9 @@ public class RabbitBinderTests extends properties.getExtension().setRequeueRejected(true); properties.getExtension().setTransacted(true); properties.getExtension().setExclusive(true); + properties.getExtension().setMissingQueuesFatal(true); + properties.getExtension().setFailedDeclarationRetryInterval(1500L); + properties.getExtension().setQueueDeclarationRetries(23); Binding consumerBinding = binder.bindConsumer("props.0", null, createBindableChannel("input", new BindingProperties()), properties); Lifecycle endpoint = extractEndpoint(consumerBinding); @@ -268,6 +271,9 @@ public class RabbitBinderTests extends assertThat(TestUtils.getPropertyValue(container, "defaultRequeueRejected", Boolean.class)).isTrue(); assertThat(TestUtils.getPropertyValue(container, "prefetchCount")).isEqualTo(1); assertThat(TestUtils.getPropertyValue(container, "txSize")).isEqualTo(1); + assertThat(TestUtils.getPropertyValue(container, "missingQueuesFatal", Boolean.class)).isTrue(); + assertThat(TestUtils.getPropertyValue(container, "failedDeclarationRetryInterval")).isEqualTo(1500L); + assertThat(TestUtils.getPropertyValue(container, "declarationRetries")).isEqualTo(23); RetryTemplate retry = TestUtils.getPropertyValue(endpoint, "retryTemplate", RetryTemplate.class); assertThat(TestUtils.getPropertyValue(retry, "retryPolicy.maxAttempts")).isEqualTo(3); assertThat(TestUtils.getPropertyValue(retry, "backOffPolicy.initialInterval")).isEqualTo(1000L); @@ -320,6 +326,7 @@ public class RabbitBinderTests extends Lifecycle endpoint = extractEndpoint(consumerBinding); SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint, "messageListenerContainer", SimpleMessageListenerContainer.class); + assertThat(TestUtils.getPropertyValue(container, "missingQueuesFatal", Boolean.class)).isFalse(); assertThat(container.isRunning()).isTrue(); consumerBinding.unbind(); assertThat(container.isRunning()).isFalse();