From 06266683147a9b03326b3277ed24113178ebdb8e Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 16 Aug 2017 11:46:14 -0400 Subject: [PATCH] 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 --- .../src/main/asciidoc/index.adoc | 3 + .../src/main/asciidoc/overview.adoc | 39 +++++++++ spring-cloud-stream-binder-rabbit/pom.xml | 18 ++++ .../rabbit/RabbitMessageChannelBinder.java | 41 ++++++++- .../binder/rabbit/RabbitBinderTests.java | 83 ++++++++++++++++++- .../binder/rabbit/RabbitTestBinder.java | 32 +++---- 6 files changed, 196 insertions(+), 20 deletions(-) diff --git a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/index.adoc b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/index.adoc index 76626b66b..3d6cf97ea 100644 --- a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/index.adoc +++ b/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[] 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 19283f93f..206cf9afa 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 @@ -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 <> 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 <>, 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 <>); you can consume these exceptions with your own Spring Integration flow. diff --git a/spring-cloud-stream-binder-rabbit/pom.xml b/spring-cloud-stream-binder-rabbit/pom.xml index f0288ba73..719d0ec8c 100644 --- a/spring-cloud-stream-binder-rabbit/pom.xml +++ b/spring-cloud-stream-binder-rabbit/pom.xml @@ -49,6 +49,24 @@ org.springframework.integration spring-integration-amqp + + 4.3.12.BUILD-SNAPSHOT + + + org.springframework.integration + spring-integration-core + 4.3.12.BUILD-SNAPSHOT + + + org.springframework.integration + spring-integration-jmx + 4.3.12.BUILD-SNAPSHOT + + + org.springframework.integration + spring-integration-test + 4.3.12.BUILD-SNAPSHOT + org.springframework.cloud 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 3a2e3fe83..4cdcd1086 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 @@ -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 producerProperties) - throws Exception { + ExtendedProducerProperties 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()); 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 1794abf4f..ef5c89d78 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 @@ -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 producerProps = createProducerProperties(); + producerProps.setErrorChannelEnabled(true); + Binding 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> 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 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); diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java index 22af545b7..d77fda360 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java +++ b/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 exchanges = new HashSet(); + 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 bindConsumer(String name, String group, MessageChannel moduleInputChannel, ExtendedConsumerProperties properties) { @@ -129,6 +126,13 @@ public class RabbitTestBinder extends AbstractTestBinder