diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java index 5525f4b5b..126e570d3 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java @@ -20,6 +20,7 @@ import java.io.IOException; import java.util.LinkedHashMap; import java.util.Map; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; import com.fasterxml.jackson.core.JsonGenerator; import com.fasterxml.jackson.databind.ObjectMapper; @@ -32,6 +33,9 @@ import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.support.DefaultSingletonBeanRegistry; +import org.springframework.cloud.function.context.FunctionCatalog; +import org.springframework.cloud.stream.config.BindingProperties; +import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.config.ConsumerEndpointCustomizer; import org.springframework.cloud.stream.config.ListenerContainerCustomizer; import org.springframework.cloud.stream.config.MessageSourceCustomizer; @@ -66,9 +70,11 @@ import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.support.ChannelInterceptor; +import org.springframework.messaging.support.ErrorMessage; import org.springframework.retry.RecoveryCallback; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; +import org.springframework.util.StringUtils; /** * {@link AbstractBinder} that serves as base class for {@link MessageChannel} binders. @@ -687,6 +693,25 @@ public abstract class AbstractMessageChannelBinder errorHandler = catalog.lookup(Consumer.class, bp.getErrorHandlerDefinition()); + if (errorHandler == null) { + logger.warn("Failed to retrieve error handling function with definition: " + bp.getErrorHandlerDefinition() + ", for binding: " + bindingName); + } + else { + SubscribableChannel functionErrorChannel = getApplicationContext().getBean(errorChannelName, SubscribableChannel.class); + functionErrorChannel.subscribe(errorMessage -> errorHandler.accept((ErrorMessage) errorMessage)); + } + } + } + /** * Build an errorChannelRecoverer that writes to a pub/sub channel for the destination * when an exception is thrown to a consumer. @@ -720,7 +745,10 @@ public abstract class AbstractMessageChannelBinder errorChannel); + + this.subscribeFunctionErrorHandler(errorChannelName, consumerProperties.getBindingName()); } + ErrorMessageSendingRecoverer recoverer; if (errorMessageStrategy == null) { recoverer = new ErrorMessageSendingRecoverer(errorChannel); diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java index 1db610e91..3fa5a0254 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java @@ -46,6 +46,11 @@ public class BindingProperties { private static final String COMMA = ","; + /** + * Function that corresponds to the error handler of the underlying function. + */ + private String errorHandlerDefinition; + /** * The physical name at the broker that the binder binds to. */ @@ -157,4 +162,12 @@ public class BindingProperties { return "BindingProperties{" + sb.toString() + "}"; } + public String getErrorHandlerDefinition() { + return errorHandlerDefinition; + } + + public void setErrorHandlerDefinition(String errorHandlerDefinition) { + this.errorHandlerDefinition = errorHandlerDefinition; + } + } diff --git a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ErrorBindingTests.java b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ErrorBindingTests.java index 3f07dd0be..92f2479b2 100644 --- a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ErrorBindingTests.java +++ b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ErrorBindingTests.java @@ -20,26 +20,17 @@ import java.util.function.Consumer; import java.util.function.Function; import org.junit.jupiter.api.Test; -import org.mockito.Mockito; -import org.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.builder.SpringApplicationBuilder; import org.springframework.cloud.stream.binder.test.InputDestination; import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; import org.springframework.context.ApplicationContext; -import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; -import org.springframework.integration.annotation.ServiceActivator; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageChannel; import org.springframework.messaging.support.GenericMessage; import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.ArgumentMatchers.isNull; /** * @author Marius Bogoevici @@ -47,25 +38,6 @@ import static org.mockito.ArgumentMatchers.isNull; */ public class ErrorBindingTests { - @SuppressWarnings({ "rawtypes", "unchecked" }) - @Test - void testErrorChannelNotBoundByDefault() { - ConfigurableApplicationContext applicationContext = SpringApplication.run( - TestProcessor.class, "--server.port=0", - "--spring.cloud.stream.default-binder=mock", - "--spring.jmx.enabled=false"); - BinderFactory binderFactory = applicationContext.getBean(BinderFactory.class); - - Binder binder = binderFactory.getBinder(null, MessageChannel.class); - - Mockito.verify(binder).bindConsumer(eq("processor-in-0"), isNull(), - any(MessageChannel.class), any(ConsumerProperties.class)); - Mockito.verify(binder).bindProducer(eq("processor-out-0"), any(MessageChannel.class), - any(ProducerProperties.class)); - Mockito.verifyNoMoreInteractions(binder); - applicationContext.close(); - } - @Test void testConfigurationWithDefaultErrorHandler() { ApplicationContext context = new SpringApplicationBuilder( @@ -73,6 +45,8 @@ public class ErrorBindingTests { ErrorBindingTests.ErrorConfigurationDefault.class)) .web(WebApplicationType.NONE) .run("--spring.cloud.stream.bindings.handle-in-0.consumer.max-attempts=1", + "--spring.cloud.function.definition=handle", + "--spring.cloud.stream.default.error-handler-definition=errorHandler", "--spring.jmx.enabled=false"); InputDestination source = context.getBean(InputDestination.class); @@ -82,16 +56,18 @@ public class ErrorBindingTests { ErrorConfigurationDefault errorConfiguration = context .getBean(ErrorConfigurationDefault.class); - assertThat(errorConfiguration.counter == 3); + assertThat(errorConfiguration.counter).isEqualTo(6); } @Test - void testConfigurationWithCustomErrorHandler() { + void testConfigurationWithBindingSpecificErrorHandler() { ApplicationContext context = new SpringApplicationBuilder( TestChannelBinderConfiguration.getCompleteConfiguration( ErrorBindingTests.ErrorConfigurationWithCustomErrorHandler.class)) .web(WebApplicationType.NONE) .run("--spring.cloud.stream.bindings.handle-in-0.consumer.max-attempts=1", + "--spring.cloud.function.definition=handle", + "--spring.cloud.stream.bindings.handle-in-0.error-handler-definition=errorHandler", "--spring.jmx.enabled=false"); InputDestination source = context.getBean(InputDestination.class); @@ -101,7 +77,7 @@ public class ErrorBindingTests { ErrorConfigurationWithCustomErrorHandler errorConfiguration = context .getBean(ErrorConfigurationWithCustomErrorHandler.class); - assertThat(errorConfiguration.counter == 6); + assertThat(errorConfiguration.counter).isEqualTo(6); } @EnableAutoConfiguration @@ -126,6 +102,13 @@ public class ErrorBindingTests { }; } + @Bean + public Consumer errorHandler() { + return v -> { + this.counter++; + }; + } + } @EnableAutoConfiguration @@ -141,11 +124,12 @@ public class ErrorBindingTests { }; } - @ServiceActivator(inputChannel = "input.anonymous.errors") - public void error(Message message) { - this.counter++; + @Bean + public Consumer errorHandler() { + return v -> { + this.counter++; + }; } - } } diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index 387666a4a..cfe0d0b94 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -1516,6 +1516,43 @@ public Function, Flux> uppercase() { By default, if no additional system-level configuration is provided, the messaging system drops the failed message. While acceptable in some cases, for most cases, it is not, and we need some recovery mechanism to avoid message loss. +==== Handle Error Messages + +In the previous section we mentioned that by default messages that resulted in error are logged and dropped. In the event you want to add some additional error handling (i.e., send notification, write to database etc), you can add a `Consumer` that is specifically designed to accept `ErrorMessage`. + +[source,java] +---- +@Bean +public Consumer myErrorHandler() { + return v -> { + // send SMS notification code + }; +} +---- + +If by accident you declare such handler as a `Function`, it will still work with the exception that nothing is going to be done with its output. However, given that such handler is still relying on functionality provided by Spring Cloud Function, you can also benefit from function composition in the event your handler has some complexity which you would like to address through function composition. + +To identify such function/consumer as an error handler all you need is to provide `error-handler-definition` property pointing to the function name - `spring.cloud.stream.bindings..error-handler-definition=myErrorHandler`. This is where you can specify function composition instruction using ‘|’ symbol as you would do with `spring.cloud.function.definition`. + +For example, for binding name `uppercase-in-0` the property would look like this: + +[source,text] +---- +spring.cloud.stream.bindings.uppercase-in-0.error-handler-definition=myErrorHandler +---- +And if you used special mapping instruction to map binding to a more readable name - `spring.cloud.stream.function.bindings.uppercase-in-0=upper`, then this property would look like this: +[source,text] +---- +spring.cloud.stream.bindings.upper.error-handler-definition=myErrorHandler`. +---- + +***Default Error Handler*** + +If you want to have a single error handler for all function beans, you can use the standard spring-cloud-stream mechanism for defining default properties `spring.cloud.stream.default.error-handler-definition=myErrorHandler` + +NOTE: When declaring function-based error handler you MUST define `spring-cloud-function-definition` to identify your bindings, since +you are effectively declaring another function that will be available in Function Catalog. So for the above case even if until declaring `myErrorHandler` function you only had a single `uppercase` function in your configuration, you now must explicitly declare `spring-cloud-function-definition=uppercase`. But as we mentioned several times before, it is a best practice and recommended approach to ALWAYS declare `spring.cloud.function.definition`. + ==== DLQ - Dead Letter Queue Perhaps the most common mechanism, DLQ allows failed messages to be sent to a special destination: the _Dead Letter Queue_.