GH-104: Missing queues fatal 'false' by default

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/104

To handle the use case of a Rabbit cluster with non-HA queues where the node hosting
a queue is down, set `missingQueuesFatal` to `false` by default.

Expose this property in the consumer config, together with the redeclaration retry properties.

Docs

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;
+	}
+
 }

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.

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);

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<MessageChannel> 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();