From 0eab8a5fa700e19c2a026722418d920955d38c7a Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 15 Aug 2017 17:11:40 -0400 Subject: [PATCH] GH-916: MC Binder Producer Error Infrastructure Initial Commit for GH-916. - register a pubsub error channel. - register and subscribe a bridge handler to bridge it to the global error channel. - pass the error channel to the implementation so it can wire it into the outbound endpoint. - destroy the infrastructure when unbinding. Javadoc Polishing Add test case. temp update to SI 4.3.12 GH-802 - Error Handling Documentation Resolves #802 --- .../spring-cloud-stream-overview.adoc | 23 +++++- spring-cloud-stream/pom.xml | 2 + .../binder/AbstractMessageChannelBinder.java | 81 ++++++++++++++++++- .../stream/binder/ProducerProperties.java | 10 +++ .../AbstractMessageChannelBinderTests.java | 45 +++++++++-- 5 files changed, 152 insertions(+), 9 deletions(-) diff --git a/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc b/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc index 0d01c0e8c..82ee1c198 100644 --- a/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc +++ b/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc @@ -447,7 +447,23 @@ Spring Cloud Stream supports publishing error messages received by the Spring In error channel. Error messages sent to the `errorChannel` can be published to a specific destination at the broker by configuring a binding for the outbound target named `error`. For example, to publish error messages to a broker destination named "myErrors", provide the following property: -`spring.cloud.stream.bindings.error.destination=myErrors` +`spring.cloud.stream.bindings.error.destination=myErrors`. + +[[binder-error-channels]] +===== Message Channel Binders and Error Channels + +Starting with _version 1.3_, some `MessageChannel` - based binders publish errors to a discrete error channel for each destination. +In addition, these error channels are bridged to the global Spring Integration `errorChannel` mentioned above. +You can therefore consume errors for specific destinations and/or for all destinations, using a standard Spring Integration flow (`IntegrationFlow`, `@ServiceActivator`, etc). + +On the consumer side, the listener thread catches any exceptions and forwards an `ErrorMessage` to the destination's error channel. +The payload of the message is a `MessagingException` with the normal `failedMessage` and `cause` properties. +Usually, the raw data received from the broker is included in a header. +For binders that support (and are configured with) a dead letter destination; a `MessagePublishingErrorHandler` is subscribed to the channel, and the raw data is forwarded to the dead letter destination. + +On the producer side; for binders that support some kind of async result after publishing messages (e.g. RabbitMQ, Kafka), you can enable an error channel by setting the `...producer.errorChannelEnabled` to `true`. +The payload of the `ErrorMessage` depends on the binder implementation but will be a `MessagingException` with the normal `failedMessage` property, as well as additional properties about the failure. +Refer to the binder documentation for complete details. ===== Using @StreamListener for Automatic Content Type Handling @@ -1246,6 +1262,11 @@ When native encoding is used, it is the responsibility of the consumer to use ap Also, when native encoding/decoding is used the `headerMode` property is ignored and headers will not be embedded into the message. + Default: `false`. +errorChannelEnabled:: + When set to `true`, if the binder supports async send results; send failures will be sent to an error channel for the destination. + See <> for more information. ++ +Default: `false`. [[dynamicdestination]] === Using dynamically bound destinations diff --git a/spring-cloud-stream/pom.xml b/spring-cloud-stream/pom.xml index efb74c64b..135797e36 100644 --- a/spring-cloud-stream/pom.xml +++ b/spring-cloud-stream/pom.xml @@ -33,10 +33,12 @@ org.springframework.integration spring-integration-core + 4.3.12.BUILD-SNAPSHOT org.springframework.integration spring-integration-jmx + 4.3.12.BUILD-SNAPSHOT org.springframework diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java index 22600d203..6100a2136 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java @@ -29,6 +29,7 @@ import org.springframework.context.Lifecycle; import org.springframework.expression.ExpressionParser; import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.channel.FixedSubscriberChannel; +import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.endpoint.EventDrivenConsumer; @@ -117,7 +118,10 @@ public abstract class AbstractMessageChannelBinder producerBinding = binder.bindProducer("bar", new DirectChannel(), - new ProducerProperties()); + ProducerProperties producerProps = new ProducerProperties(); + producerProps.setErrorChannelEnabled(true); + Binding producerBinding = binder.bindProducer("bar", new DirectChannel(), producerProps); DirectFieldAccessor producerBindingAccessor = new DirectFieldAccessor(producerBinding); Object messageHandler = producerBindingAccessor.getPropertyValue("lifecycle"); Mockito.verify((Lifecycle) messageHandler).start(); Mockito.verify((InitializingBean) messageHandler).afterPropertiesSet(); Mockito.verifyNoMoreInteractions(messageHandler); + assertThat(context.containsBean("bar.errors")).isTrue(); + assertThat(context.containsBean("bar.errors.bridge")).isTrue(); producerBinding.unbind(); Mockito.verify((Lifecycle) messageHandler).stop(); Mockito.verify((DisposableBean) messageHandler).destroy(); Mockito.verifyNoMoreInteractions(messageHandler); + assertThat(context.containsBean("bar.errors")).isFalse(); + assertThat(context.containsBean("bar.errors.bridge")).isFalse(); } @Test @@ -165,12 +170,21 @@ public class AbstractMessageChannelBinderTests { } }).given(this.provisioningProvider).provisionConsumerDestination(anyString(), anyString(), - Matchers.any(ConsumerProperties.class)); + any(ConsumerProperties.class)); + willAnswer(new Answer() { + + @Override + public SimpleProducerDestination answer(final InvocationOnMock invocation) throws Throwable { + return new SimpleProducerDestination(invocation.getArgumentAt(0, String.class)); + } + + }).given(this.provisioningProvider).provisionProducerDestination(anyString(), + any(ProducerProperties.class)); } @Override protected MessageHandler createProducerMessageHandler(ProducerDestination destination, - ProducerProperties producerProperties) throws Exception { + ProducerProperties producerProperties, MessageChannel errorChannel) throws Exception { MessageHandler mock = Mockito.mock(MessageHandler.class, Mockito.withSettings() .extraInterfaces(Lifecycle.class, InitializingBean.class, DisposableBean.class)); return mock; @@ -214,4 +228,25 @@ public class AbstractMessageChannelBinderTests { } + private static class SimpleProducerDestination implements ProducerDestination { + + private final String name; + + + SimpleProducerDestination(String name) { + this.name = name; + } + + @Override + public String getName() { + return this.name; + } + + @Override + public String getNameForPartition(int partition) { + return getName() + partition; + } + + } + }