SCSt-GH-916: Configure Producer Error Channel

Requires: https://github.com/spring-cloud/spring-cloud-stream/pull/1039

Publish errors (returned and nack'd messages) to the error channel.

Add docs

Test Polishing

spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/index.adoc

@@ -22,13 +22,16 @@ Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinat
 // ======================================================================================
 
 = Reference Guide
+
 include::overview.adoc[]
+
 include::dlq.adoc[]
 
 
 = Appendices
 [appendix]
 include::building.adoc[]
+
 [appendix]
 include::contributing.adoc[]
 

spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc

@@ -502,3 +502,42 @@ public class XDeathApplication {
 ----
 
 Notice that the count property in the `x-death` header is a `Long`.
+
+[[rabbit-error-channels]]
+== Error Channels
+
+Starting with _version 1.3_, the binder unconditionally sends exceptions to an error channel for each consumer destination, and can be configured to send async producer send failures to an error channel too.
+See <<binder-error-channels>> for more information.
+
+With rabbitmq, there are two types of send failures:
+
+* returned messages
+* negatively acknowledged https://www.rabbitmq.com/confirms.html[Publisher Confirms]
+
+The latter is rare; quoting the RabbitMQ documentation "[A nack] will only be delivered if an internal error occurs in the Erlang process responsible for a queue.".
+
+As well as enabling producer error channels as described in <<binder-error-channels>>, the RabbitMQ binder will only send messages to the channels if the connection factory is appropriately configured:
+
+* `ccf.setPublisherConfirms(true);`
+* `ccf.setPublisherReturns(true);`
+
+When using spring boot configuration for the connection factory, set properties:
+
+* `spring.rabbitmq.publisher-confirms`
+* `spring.rabbitmq.publisher-returns`
+
+The payload of the `ErrorMessage` for a returned message is a `ReturnedAmqpMessageException` with properties:
+
+* `failedMessage` - the spring-messaging `Message<?>` that failed to be sent.
+* `amqpMessage` - the raw spring-amqp `Message`
+* `replyCode` - an integer value indicating the reason for the failure (e.g. 312 - No route)
+* `replyText` - a text value indicating the reason for the failure e.g. `NO_ROUTE`.
+* `exchange` - the exchange to which the message was published.
+* `routingKey` - the routing key used when the message was published.
+
+For negatively acknowledged confirms, the payload is a `NackedAmqpMessageException` with properties:
+
+* `failedMessage` - the spring-messaging `Message<?>` that failed to be sent.
+* `nackReason` - a reason (if available; you may need to examine the broker logs for more information).
+
+There is no automatic handling of these exceptions (such as sending to a <<rabbit-dlq-processing, Dead-Letter queue>>); you can consume these exceptions with your own Spring Integration flow.

spring-cloud-stream-binder-rabbit/pom.xml

@@ -49,6 +49,24 @@
 		<dependency>
 			<groupId>org.springframework.integration</groupId>
 			<artifactId>spring-integration-amqp</artifactId>
+<!-- begin temporary -->
+			<version>4.3.12.BUILD-SNAPSHOT</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.integration</groupId>
+			<artifactId>spring-integration-core</artifactId>
+			<version>4.3.12.BUILD-SNAPSHOT</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.integration</groupId>
+			<artifactId>spring-integration-jmx</artifactId>
+			<version>4.3.12.BUILD-SNAPSHOT</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.integration</groupId>
+			<artifactId>spring-integration-test</artifactId>
+			<version>4.3.12.BUILD-SNAPSHOT</version>
+<!-- end temporary -->
 		</dependency>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>

spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java

@@ -27,6 +27,7 @@ import org.springframework.amqp.AmqpRejectAndDontRequeueException;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.connection.LocalizedQueueConnectionFactory;
 import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate;
@@ -63,6 +64,7 @@ import org.springframework.integration.amqp.support.AmqpMessageHeaderErrorMessag
 import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.core.MessageProducer;
+import org.springframework.integration.support.DefaultErrorMessageStrategy;
 import org.springframework.integration.support.ErrorMessageStrategy;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
@@ -194,12 +196,13 @@ public class RabbitMessageChannelBinder
 
 	@Override
 	protected MessageHandler createProducerMessageHandler(final ProducerDestination producerDestination,
-														ExtendedProducerProperties<RabbitProducerProperties> producerProperties)
-			throws Exception {
+			ExtendedProducerProperties<RabbitProducerProperties> producerProperties, MessageChannel errorChannel)
+					throws Exception {
 		String prefix = producerProperties.getExtension().getPrefix();
 		String exchangeName = producerDestination.getName();
 		String destination = StringUtils.isEmpty(prefix) ? exchangeName : exchangeName.substring(prefix.length());
-		final AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(buildRabbitTemplate(producerProperties.getExtension()));
+		final AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(
+				buildRabbitTemplate(producerProperties.getExtension(), errorChannel != null));
 		endpoint.setExchangeName(producerDestination.getName());
 		RabbitProducerProperties extendedProperties = producerProperties.getExtension();
 		String routingKeyExpression = extendedProperties.getRoutingKeyExpression();
@@ -230,10 +233,39 @@ public class RabbitMessageChannelBinder
 		endpoint.setHeaderMapper(mapper);
 		endpoint.setDefaultDeliveryMode(extendedProperties.getDeliveryMode());
 		endpoint.setBeanFactory(this.getBeanFactory());
+		if (errorChannel != null) {
+			checkConnectionFactoryIsErrorCapable();
+			endpoint.setReturnChannel(errorChannel);
+			endpoint.setConfirmNackChannel(errorChannel);
+			endpoint.setConfirmCorrelationExpressionString("#root");
+			endpoint.setErrorMessageStrategy(new DefaultErrorMessageStrategy());
+		}
 		endpoint.afterPropertiesSet();
 		return endpoint;
 	}
 
+	private void checkConnectionFactoryIsErrorCapable() {
+		if (!(this.connectionFactory instanceof CachingConnectionFactory)) {
+			logger.warn("Unknown connection factory type, cannot determine error capabilities: "
+					+ this.connectionFactory.getClass());
+		}
+		else {
+			CachingConnectionFactory ccf = (CachingConnectionFactory) this.connectionFactory;
+			if (!ccf.isPublisherConfirms() && !ccf.isPublisherReturns()) {
+				logger.warn("Producer error channel is enabled, but the connection factory is not configured for "
+						+ "returns or confirms; the error channel will receive no messages");
+			}
+			else if (!ccf.isPublisherConfirms()) {
+				logger.info("Producer error channel is enabled, but the connection factory is only configured to "
+						+ "handle returned messages; negative acks will not be reported");
+			}
+			else if (!ccf.isPublisherReturns()) {
+				logger.info("Producer error channel is enabled, but the connection factory is only configured to "
+						+ "handle negatively acked messages; returned messages will not be reported");
+			}
+		}
+	}
+
 	private String buildPartitionRoutingExpression(String expressionRoot, boolean rootIsExpression) {
 		return rootIsExpression
 				? expressionRoot + " + '-' + headers['" + BinderHeaders.PARTITION_HEADER + "']"
@@ -390,7 +422,7 @@ public class RabbitMessageChannelBinder
 		provisioningProvider.cleanAutoDeclareContext(consumerDestination.getName());
 	}
 
-	private RabbitTemplate buildRabbitTemplate(RabbitProducerProperties properties) {
+	private RabbitTemplate buildRabbitTemplate(RabbitProducerProperties properties, boolean mandatory) {
 		RabbitProperties rabbitProperties = null;
 		try {
 			rabbitProperties = getApplicationContext().getBean(RabbitProperties.class);
@@ -416,6 +448,7 @@ public class RabbitMessageChannelBinder
 			rabbitTemplate.setBeforePublishPostProcessors(this.compressingPostProcessor);
 		}
 		rabbitTemplate.setChannelTransacted(properties.isTransacted());
+		rabbitTemplate.setMandatory(mandatory); // returned messages
 		if (rabbitProperties != null && rabbitProperties.getTemplate().getRetry().isEnabled()) {
 			Retry retry = rabbitProperties.getTemplate().getRetry();
 			RetryPolicy retryPolicy = new SimpleRetryPolicy(retry.getMaxAttempts());

spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java

@@ -16,6 +16,7 @@
 
 package org.springframework.cloud.stream.binder.rabbit;
 
+import java.lang.reflect.Constructor;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
@@ -34,6 +35,7 @@ import org.mockito.ArgumentCaptor;
 
 import org.springframework.amqp.AmqpIOException;
 import org.springframework.amqp.core.AcknowledgeMode;
+import org.springframework.amqp.core.AmqpTemplate;
 import org.springframework.amqp.core.BindingBuilder;
 import org.springframework.amqp.core.DirectExchange;
 import org.springframework.amqp.core.Exchange;
@@ -70,8 +72,12 @@ import org.springframework.context.ApplicationContext;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.Lifecycle;
 import org.springframework.expression.spel.standard.SpelExpression;
+import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
+import org.springframework.integration.amqp.support.NackedAmqpMessageException;
+import org.springframework.integration.amqp.support.ReturnedAmqpMessageException;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
@@ -79,13 +85,16 @@ import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.SubscribableChannel;
+import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.retry.support.RetryTemplate;
+import org.springframework.util.ReflectionUtils;
 
 import com.rabbitmq.http.client.domain.QueueInfo;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.fail;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -163,6 +172,77 @@ public class RabbitBinderTests extends
 		consumerBinding.unbind();
 	}
 
+	@Test
+	public void testProducerErrorChannel() throws Exception {
+		RabbitTestBinder binder = getBinder();
+		CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource();
+		ccf.setPublisherReturns(true);
+		ccf.setPublisherConfirms(true);
+		ccf.resetConnection();
+		DirectChannel moduleOutputChannel = createBindableChannel("output", new BindingProperties());
+		ExtendedProducerProperties<RabbitProducerProperties> producerProps = createProducerProperties();
+		producerProps.setErrorChannelEnabled(true);
+		Binding<MessageChannel> producerBinding = binder.bindProducer("ec.0", moduleOutputChannel, producerProps);
+		final Message<?> message = MessageBuilder.withPayload("bad").setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar")
+				.build();
+		SubscribableChannel ec = binder.getApplicationContext().getBean("ec.0.errors", SubscribableChannel.class);
+		final AtomicReference<Message<?>> errorMessage = new AtomicReference<>();
+		final CountDownLatch latch = new CountDownLatch(2);
+		ec.subscribe(new MessageHandler() {
+
+			@Override
+			public void handleMessage(Message<?> message) throws MessagingException {
+				errorMessage.set(message);
+				latch.countDown();
+			}
+
+		});
+		SubscribableChannel globalEc = binder.getApplicationContext()
+				.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, SubscribableChannel.class);
+		globalEc.subscribe(new MessageHandler() {
+
+			@Override
+			public void handleMessage(Message<?> message) throws MessagingException {
+				latch.countDown();
+			}
+
+		});
+		moduleOutputChannel.send(message);
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(errorMessage.get()).isInstanceOf(ErrorMessage.class);
+		assertThat(errorMessage.get().getPayload()).isInstanceOf(ReturnedAmqpMessageException.class);
+		ReturnedAmqpMessageException exception = (ReturnedAmqpMessageException) errorMessage.get().getPayload();
+		assertThat(exception.getReplyCode()).isEqualTo(312);
+		assertThat(exception.getReplyText()).isEqualTo("NO_ROUTE");
+
+		AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(producerBinding, "lifecycle",
+				AmqpOutboundEndpoint.class);
+		assertThat(TestUtils.getPropertyValue(endpoint, "confirmCorrelationExpression.expression"))
+				.isEqualTo("#root");
+		class WrapperAccessor extends AmqpOutboundEndpoint {
+
+			public WrapperAccessor(AmqpTemplate amqpTemplate) {
+				super(amqpTemplate);
+			}
+
+			public CorrelationDataWrapper getWrapper() throws Exception {
+				Constructor<CorrelationDataWrapper> constructor = CorrelationDataWrapper.class.getDeclaredConstructor(
+						String.class, Object.class, Message.class);
+				ReflectionUtils.makeAccessible(constructor);
+				return constructor.newInstance(null, message, message);
+			}
+
+		}
+		endpoint.confirm(new WrapperAccessor(mock(AmqpTemplate.class)).getWrapper(), false, "Mock NACK");
+		assertThat(errorMessage.get()).isInstanceOf(ErrorMessage.class);
+		assertThat(errorMessage.get().getPayload()).isInstanceOf(NackedAmqpMessageException.class);
+		NackedAmqpMessageException nack = (NackedAmqpMessageException) errorMessage.get().getPayload();
+		assertThat(nack.getNackReason()).isEqualTo("Mock NACK");
+		assertThat(nack.getCorrelationData()).isEqualTo(message);
+		assertThat(nack.getFailedMessage()).isEqualTo(message);
+		producerBinding.unbind();
+	}
+
 	@Test
 	public void testConsumerProperties() throws Exception {
 		RabbitTestBinder binder = getBinder();
@@ -1103,8 +1183,7 @@ public class RabbitBinderTests extends
 	@Test
 	public void testBadUserDeclarationsFatal() {
 		RabbitTestBinder binder = getBinder();
-		ConfigurableApplicationContext context = TestUtils.getPropertyValue(binder, "binder.applicationContext",
-				ConfigurableApplicationContext.class);
+		ConfigurableApplicationContext context = binder.getApplicationContext();
 		ConfigurableListableBeanFactory bf = context.getBeanFactory();
 		bf.registerSingleton("testBadUserDeclarationsFatal", new Queue("testBadUserDeclarationsFatal", false));
 		bf.registerSingleton("binder", binder);

spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java

@@ -30,12 +30,11 @@ import org.springframework.cloud.stream.binder.rabbit.properties.RabbitCommonPro
 import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties;
 import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties;
 import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner;
-import org.springframework.context.support.GenericApplicationContext;
-import org.springframework.integration.channel.PublishSubscribeChannel;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.codec.kryo.PojoCodec;
-import org.springframework.integration.context.IntegrationContextUtils;
+import org.springframework.integration.config.EnableIntegration;
 import org.springframework.messaging.MessageChannel;
-import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 
 /**
  * Test support class for {@link RabbitMessageChannelBinder}.
@@ -55,27 +54,25 @@ public class RabbitTestBinder extends AbstractTestBinder<RabbitMessageChannelBin
 
 	private final Set<String> exchanges = new HashSet<String>();
 
+	private final AnnotationConfigApplicationContext applicationContext;
+
 	public RabbitTestBinder(ConnectionFactory connectionFactory, RabbitProperties rabbitProperties) {
 		this(connectionFactory, new RabbitMessageChannelBinder(connectionFactory, rabbitProperties,
 				new RabbitExchangeQueueProvisioner(connectionFactory)));
 	}
 
 	public RabbitTestBinder(ConnectionFactory connectionFactory, RabbitMessageChannelBinder binder) {
-		GenericApplicationContext context = new GenericApplicationContext();
-		ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
-		scheduler.setPoolSize(1);
-		scheduler.afterPropertiesSet();
-		context.getBeanFactory().registerSingleton(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, scheduler);
-		PublishSubscribeChannel errorChannel = new PublishSubscribeChannel();
-		context.getBeanFactory().initializeBean(errorChannel, IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME);
-		context.getBeanFactory().registerSingleton(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, errorChannel);
-		context.refresh();
-		binder.setApplicationContext(context);
+		this.applicationContext = new AnnotationConfigApplicationContext(Config.class);
+		binder.setApplicationContext(this.applicationContext);
 		binder.setCodec(new PojoCodec());
 		this.setBinder(binder);
 		this.rabbitAdmin = new RabbitAdmin(connectionFactory);
 	}
 
+	public AnnotationConfigApplicationContext getApplicationContext() {
+		return this.applicationContext;
+	}
+
 	@Override
 	public Binding<MessageChannel> bindConsumer(String name, String group, MessageChannel moduleInputChannel,
 			ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
@@ -129,6 +126,13 @@ public class RabbitTestBinder extends AbstractTestBinder<RabbitMessageChannelBin
 		for (String prefix : this.prefixes) {
 			this.rabbitAdmin.deleteExchange(prefix + "DLX");
 		}
+		this.applicationContext.close();
+	}
+
+	@Configuration
+	@EnableIntegration
+	static class Config {
+
 	}
 
 }