diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index 88064b287..dc9bf9be1 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -1347,26 +1347,23 @@ NOTE: If you need to support dynamic destinations with multiple binder types, us [[spring-cloud-stream-overview-error-handling]] === Error Handling -Errors happen, and Spring Cloud Stream provides several flexible mechanism to handle them by delegating to the -binder (re-queue, DL, and others). Note that the techniques are dependent on binder implementation and the +In this section we'll explain the general idea behind error handling mechanisms provided by the framework. +We'll be using Rabbit binder as an example, since individual binders define different set +of properties for certain supported mechanisms specific to underlying broker capabilities (such as Kafka bindre). + +Errors happen, and Spring Cloud Stream provides several flexible mechanisms to deal with them +(re-queue, DL, and others). Note that the techniques are dependent on binder implementation and the capability of the underlying messaging middleware. -Also, for non-reactive functions, Spring Cloud Stream uses the https://github.com/spring-projects/spring-retry[Spring Retry] library to -facilitate successful message processing. See <> for more details. -However, when all fails, the exceptions thrown by the message handlers are propagated back to the binder. +Whenever there is an exception during message processing, the framework will make several attempts at re-trying +the same message (3 by default). For that, the framework uses https://github.com/spring-projects/spring-retry[Spring Retry] library +(for imperative functions and standard message handlers) and `retryBackoff` capabilities of the reactive API (for reactive +functions). -Binder-level error handling implies that the errors are communicated back to the messaging system, but given -that not every messaging system is the same, the capabilities may differ from binder to binder, so refer to -individual binder's documentation for more details. - -That said, in this section we explain the general idea behind binder level error handling and use Rabbit binder as an example. -NOTE: Kafka binder provides similar -support, although some configuration properties do differ. - -Whenever handler (function) throws and exception, it is propagated to the binder, and the binder subsequently propagates +Whenever handler (function) throws and exception, it is propagated back to the binder, and the binder subsequently propagates the error back to the messaging system. Depending on the capabilities of the messaging system such system may _drop_ the message, _re-queue_ the message for re-processing or _send the failed message to DLQ_. -Both Rabbit and Kafka support these concepts. However, other binders may not, so refer to your individual binder’s documentation for details on supported binder-level +Both Rabbit and Kafka support these concepts. However, other binders may not, so refer to your individual binder’s documentation for details on supported error-handling options. ==== Drop Failed Messages @@ -1376,7 +1373,7 @@ While acceptable in some cases, for most cases, it is not, and we need some reco ==== DLQ - Dead Letter Queue -DLQ allows failed messages to be sent to a special destination: - _Dead Letter Queue_. +Perhaps the most common mechanism, DLQ allows failed messages to be sent to a special destination: - _Dead Letter Queue_. When configured, failed messages are sent to this destination for subsequent re-processing or auditing and reconciliation. @@ -1385,38 +1382,40 @@ Consider the following example: [source,java] ---- @SpringBootApplication -public class ErrorStreamApplication { +public class SimpleStreamApplication { - public static void main(String[] args) { - SpringApplication.run(ErrorStreamApplication.class, - "--spring.cloud.stream.bindings.uppercase-in-0.group=myGroup", - "--spring.cloud.stream.rabbit.bindings.uppercase-in-0.consumer.auto-bind-dlq=true"); + public static void main(String[] args) throws Exception { + SpringApplication.run(SimpleStreamApplication.class, + "--spring.cloud.function.definition=uppercase", + "--spring.cloud.stream.bindings.uppercase-in-0.destination=uppercase", + "--spring.cloud.stream.bindings.uppercase-in-0.group=myGroup", + "--spring.cloud.stream.rabbit.bindings.uppercase-in-0.consumer.auto-bind-dlq=true" + ); } @Bean - public Function uppercase() { - return value -> {throw new RuntimeException("Intentional")}; + public Function uppercase() { + return personIn -> { + throw new RuntimeException("intentional"); + }); + }; } } ---- -Keep in mind that, in the preceding example `uppercase-in-0` corresponds to the name of the input destination binding. -The `consumer` indicates that it is a consumer property and `auto-bind-dlq` instructs the binder to configure DLQ for -`uppercase-in-0` destination, which results in an additional Rabbit queue named `uppercase-in-0.myGroup.dlq`. +As a reminder, in this example `uppercase-in-0` segment of the property corresponds to the name of the input destination binding. +The `consumer` segment indicates that it is a consumer property. + +NOTE: When using DLQ, at least `group` property must be provided for proper naming of the DLQ destination. However `group` often used together +with `destination` property, as in our example. + + +Aside from some standard properties we also set the `auto-bind-dlq` to instruct the binder to create and configure DLQ destination for +`uppercase-in-0` binding which corresponds to `uppercase` destination (see corresponding property), which results in an additional Rabbit queue named `uppercase.myGroup.dlq` (see Kafka documentation for Kafka specific DLQ properties). Once configured, all failed messages are routed to this destination preserving the original message for further actions. -However, one thing you may have noticed is that there is limited information on the original issue or the cause of the error. -For example, you do not see a stack trace corresponding to the original error. -To get more relevant information about the original error, additionally you must set `republish-to-dlq` property: - -[source,text] ----- ---spring.cloud.stream.rabbit.bindings.uppercase-in-0.consumer.republish-to-dlq=true ----- - -Doing so forces the internal error handler to intercept the error message and add additional information to it before publishing it to DLQ. -Once configured, you can see that the error message contains more information relevant to the original error, as follows: +And you can see that the error message contains more information relevant to the original error, as follows: [source,text] ---- @@ -1452,7 +1451,10 @@ In the preceding example, the `max-attempts` set to 1 essentially disabling inte Once set, the failed message is resubmitted to the same handler and loops continuously or until the handler throws `AmqpRejectAndDontRequeueException` essentially allowing you to build your own re-try logic within the handler itself. -==== Retry Template +==== Retry Template and retryBackoff + +In this section we cover configuration properties relevant to configuration of retry capabilities. +Given that we use two different mechanisms for imperative and reactive handlers (RetryTemplate and retryBackoff), properties that corresponds to both will be identified as such. The `RetryTemplate` is part of the https://github.com/spring-projects/spring-retry[Spring Retry] library. While it is out of scope of this document to cover all of the capabilities of the `RetryTemplate`, we will mention the following consumer properties that are specifically related to @@ -1461,15 +1463,15 @@ the `RetryTemplate`: maxAttempts:: The number of attempts to process the message. + -Default: 3. +Default: 3. - Applies to 'retryBackoff' backOffInitialInterval:: The backoff initial interval on retry. + -Default 1000 milliseconds. +Default 1000 milliseconds. - Applies to 'retryBackoff' backOffMaxInterval:: The maximum backoff interval. + -Default 10000 milliseconds. +Default 10000 milliseconds. - Applies to 'retryBackoff' backOffMultiplier:: The backoff multiplier. + diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java index b506b7236..f715609a1 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java @@ -34,7 +34,6 @@ import java.util.stream.Stream; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.reactivestreams.Publisher; -import reactor.core.publisher.EmitterProcessor; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.MonoSink; @@ -83,6 +82,7 @@ import org.springframework.core.annotation.AnnotationUtils; import org.springframework.core.env.Environment; import org.springframework.core.type.MethodMetadata; import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.channel.MessageChannelReactiveUtils; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlowBuilder; import org.springframework.integration.dsl.IntegrationFlows; @@ -93,6 +93,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.SubscribableChannel; +import org.springframework.messaging.support.ErrorMessage; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; import org.springframework.util.ObjectUtils; @@ -133,8 +134,6 @@ public class FunctionConfiguration { serviceProperties, dynamicDestinationResolver); } - - /* * Binding initializer responsible only for Suppliers */ @@ -364,7 +363,7 @@ public class FunctionConfiguration { Publisher[] inputPublishers = inputBindingNames.stream().map(inputBindingName -> { SubscribableChannel inputChannel = this.context.getBean(inputBindingName, SubscribableChannel.class); - return this.enhancePublisher(this.convertToPublisher(inputChannel), inputBindingName); + return MessageChannelReactiveUtils.toPublisher(inputChannel); }).toArray(Publisher[]::new); @@ -424,10 +423,11 @@ public class FunctionConfiguration { if (FunctionTypeUtils.isReactive(FunctionTypeUtils.getInputType(functionType, 0)) && StringUtils.hasText(outputChannelName)) { MessageChannel outputChannel = context.getBean(outputChannelName, MessageChannel.class); - SubscribableChannel subscribeChannel = (SubscribableChannel) inputChannel; - Publisher publisher = this.enhancePublisher(this.convertToPublisher(inputChannel), - ((DirectWithAttributesChannel) inputChannel).getBeanName()); - this.subscribeToInput(function, publisher, message -> outputChannel.send((Message) message)); + + + Publisher> publisher = MessageChannelReactiveUtils.toPublisher(inputChannel); + String bindingName = ((DirectWithAttributesChannel) inputChannel).getBeanName(); + this.subscribeToInput(function, bindingName, publisher, message -> outputChannel.send((Message) message)); } else { String inputChannelName = ((AbstractMessageChannel) inputChannel).getBeanName(); @@ -467,38 +467,40 @@ public class FunctionConfiguration { return handler; } - /* - * Enhance publisher to add error handling, retries etc. - */ @SuppressWarnings({ "unchecked", "rawtypes" }) - private Publisher enhancePublisher(Publisher publisher, String bindingName) { - Flux flux = Flux.from(publisher) + private void subscribeToInput(Function function, String bindingName, Publisher publisher, Consumer outputProcessor) { + + Flux inputPublisher = Flux.from(publisher); + + AtomicReference> originalMessageRef = new AtomicReference<>(); + AtomicReference consumerPropertiesRef = new AtomicReference<>(); + AtomicReference bindingErrorChannelRef = new AtomicReference<>(context.getBean("nullChannel", MessageChannel.class)); + + Flux result = inputPublisher + .switchOnFirst((x, message) -> { + consumerPropertiesRef.set(this.serviceProperties.getBindings().get(bindingName).getConsumer()); + String destination = serviceProperties.getBindings().get(bindingName).getDestination(); + String group = serviceProperties.getBindings().get(bindingName).getGroup(); + String bindingErrorChannelName = destination + "." + group + ".errors"; + if (context.containsBean(bindingErrorChannelName)) { + bindingErrorChannelRef.set(context.getBean(bindingErrorChannelName, MessageChannel.class)); + } + return message; + }) .concatMap(message -> { - ConsumerProperties consumerProperties = this.serviceProperties.getBindings().get(bindingName).getConsumer(); - return Flux.just(message) - .doOnError(e -> { - e.printStackTrace(); - }) - .retryBackoff( - consumerProperties.getMaxAttempts(), - Duration.ofMillis(consumerProperties.getBackOffInitialInterval()), - Duration.ofMillis(consumerProperties.getBackOffMaxInterval()) - ) + return Flux.just(message).doOnNext(originalMessageRef::set) + .transform((Function, Flux>) function) + .retryBackoff(consumerPropertiesRef.get().getMaxAttempts(), + Duration.ofMillis(consumerPropertiesRef.get().getBackOffInitialInterval()), + Duration.ofMillis(consumerPropertiesRef.get().getBackOffMaxInterval())) .onErrorResume(e -> { - e.printStackTrace(); + bindingErrorChannelRef.get() + .send(new ErrorMessage((Throwable) e, originalMessageRef.get().getHeaders(), originalMessageRef.get())); return Mono.empty(); }); - }); - return flux; - } - - @SuppressWarnings({ "unchecked", "rawtypes" }) - private void subscribeToInput(Function function, Publisher publisher, Consumer outputProcessor) { - Function, Flux> functionInvoker = function; - Flux inputPublisher = Flux.from(publisher); - subscribeToOutput(outputProcessor, functionInvoker.apply((Flux) inputPublisher)).subscribe(); + subscribeToOutput(outputProcessor, result).subscribe(); } @SuppressWarnings("rawtypes") @@ -544,13 +546,6 @@ public class FunctionConfiguration { && ((BindableFunctionProxyFactory) bindableProxyFactory).isMultiple(); } - private Publisher> convertToPublisher(SubscribableChannel inputChannel) { - EmitterProcessor> publisher = EmitterProcessor.create(1); - inputChannel.subscribe(message -> { - publisher.onNext(message); - }); - return publisher; - } } /**