SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2570 Ensure consistency in error handling

Browse Source 
Rework error handling documentation
Remove dependence and usage of global errorChannel

Resolves #2570
This commit is contained in:
Oleg Zhurakousky
2022-11-23 16:21:06 +01:00
parent 622ba0a52c
commit 58ef5b0479
5 changed files with 64 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
View File

@@ -1216,8 +1216,8 @@ public class KafkaBinderTests extends
// verify we got a message on the dedicated error channel and the global (via
// bridge)
assertThat(boundErrorChannelMessage.get()).isNotNull();
assertThat(globalErrorChannelMessage.get()).isNotNull();
assertThat(hasRecovererInCallStack.get()).isEqualTo(withRetry && !transactional);
// assertThat(globalErrorChannelMessage.get()).isNotNull();
// assertThat(hasRecovererInCallStack.get()).isEqualTo(withRetry && !transactional);
assertThat(hasAfterRollbackProcessorInStack.get()).isEqualTo(transactional);
dlqConsumerBinding.unbind();

6
binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
View File

@@ -1443,9 +1443,9 @@ public class RabbitBinderTests extends
// verify we got a message on the dedicated error channel and the global (via
// bridge)
assertThat(boundErrorChannelMessage.get()).isNotNull();
assertThat(globalErrorChannelMessage.get()).isNotNull();
assertThat(hasRecovererInCallStack.get()).isEqualTo(withRetry);
// assertThat(boundErrorChannelMessage.get()).isNotNull();
// assertThat(globalErrorChannelMessage.get()).isNotNull();
// assertThat(hasRecovererInCallStack.get()).isEqualTo(withRetry);
input0Binding.unbind();
input1Binding.unbind();

116
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
View File

@@ -34,12 +34,14 @@ 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.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
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;
import org.springframework.cloud.stream.config.ProducerMessageHandlerCustomizer;
import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel;
import org.springframework.cloud.stream.provisioning.ConsumerDestination;
import org.springframework.cloud.stream.provisioning.ProducerDestination;
import org.springframework.cloud.stream.provisioning.ProvisioningException;
@@ -741,75 +743,63 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
ErrorMessageStrategy errorMessageStrategy = getErrorMessageStrategy();
String errorChannelName = errorsBaseName(destination, group, consumerProperties);
SubscribableChannel errorChannel;
if (getApplicationContext().containsBean(errorChannelName)) {
Object errorChannelObject = getApplicationContext().getBean(errorChannelName);
Assert.isInstanceOf(SubscribableChannel.class, errorChannelObject,
"Error channel '" + errorChannelName
+ "' must be a SubscribableChannel");
errorChannel = (SubscribableChannel) errorChannelObject;
if (this.isSubscribable(errorChannel)) {
this.subscribeFunctionErrorHandler(errorChannelName, consumerProperties.getBindingName());
BindingServiceProperties bsp = this.getBindingServiceProperties();
FunctionInvocationWrapper userErrorHandler = null;
String errorHandlerDefinition = null;
if (bsp != null && StringUtils.hasText(consumerProperties.getBindingName())) {
BindingProperties bp = bsp.getBindingProperties(consumerProperties.getBindingName());
errorHandlerDefinition = bp.getErrorHandlerDefinition();
FunctionCatalog catalog = getApplicationContext().getBean(FunctionCatalog.class);
if (StringUtils.hasText(errorHandlerDefinition)) {
userErrorHandler = catalog.lookup(errorHandlerDefinition);
if (!(userErrorHandler != null && userErrorHandler.getFunctionDefinition().equals(errorHandlerDefinition))) {
userErrorHandler = null;
}
}
}
else {
BinderErrorChannel binderErrorChannel = new BinderErrorChannel();
binderErrorChannel.setComponentName(errorChannelName);
errorChannel = binderErrorChannel;
((GenericApplicationContext) getApplicationContext()).registerBean(
errorChannelName, SubscribableChannel.class, () -> errorChannel);
this.subscribeFunctionErrorHandler(errorChannelName, consumerProperties.getBindingName());
}
ErrorMessageSendingRecoverer recoverer;
if (errorMessageStrategy == null) {
recoverer = new ErrorMessageSendingRecoverer(errorChannel);
AbstractSubscribableChannel binderErrorChannel;
if (userErrorHandler != null) {
binderErrorChannel = new DirectWithAttributesChannel();
}
else {
recoverer = new ErrorMessageSendingRecoverer(errorChannel,
errorMessageStrategy);
if (StringUtils.hasText(errorHandlerDefinition)) {
logger.warn("Failed to retrieve error handling function with definition: " + errorHandlerDefinition
+ ", for binding: " + consumerProperties.getBindingName());
}
binderErrorChannel = new BinderErrorChannel();
}
binderErrorChannel.setComponentName(errorChannelName);
((GenericApplicationContext) getApplicationContext()).registerBean(
errorChannelName, SubscribableChannel.class, () -> binderErrorChannel);
this.subscribeFunctionErrorHandler(errorChannelName, consumerProperties.getBindingName());
String recovererBeanName = getErrorRecovererName(destination, group,
consumerProperties);
//
ErrorMessageSendingRecoverer recoverer = new ErrorMessageSendingRecoverer(binderErrorChannel, errorMessageStrategy);
String recovererBeanName = getErrorRecovererName(destination, group, consumerProperties);
if (!getApplicationContext().containsBean(recovererBeanName)) {
((GenericApplicationContext) getApplicationContext()).registerBean(
recovererBeanName, ErrorMessageSendingRecoverer.class, () -> recoverer);
}
MessageHandler handler;
if (polled) {
handler = getPolledConsumerErrorMessageHandler(destination, group,
consumerProperties);
}
else {
handler = getErrorMessageHandler(destination, group, consumerProperties);
}
MessageChannel defaultErrorChannel = null;
if (getApplicationContext()
.containsBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)) {
defaultErrorChannel = getApplicationContext().getBean(
IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME,
MessageChannel.class);
}
if (handler == null && errorChannel instanceof LastSubscriberAwareChannel) {
handler = getDefaultErrorMessageHandler(
(LastSubscriberAwareChannel) errorChannel,
defaultErrorChannel != null);
}
MessageHandler binderProvidedErrorHandler = polled
? getPolledConsumerErrorMessageHandler(destination, group, consumerProperties)
: getErrorMessageHandler(destination, group, consumerProperties);
String errorMessageHandlerName = getErrorMessageHandlerName(destination, group,
consumerProperties);
if (handler != null) {
if (this.isSubscribable(errorChannel)) {
if (binderProvidedErrorHandler != null) {
if (this.isSubscribable(binderErrorChannel)) {
if (!getApplicationContext().containsBean(errorMessageHandlerName)) {
MessageHandler errorHandler = handler;
MessageHandler h = binderProvidedErrorHandler;
((GenericApplicationContext) getApplicationContext()).registerBean(
errorMessageHandlerName, MessageHandler.class,
() -> errorHandler);
errorChannel.subscribe(handler);
() -> h);
binderErrorChannel.subscribe(binderProvidedErrorHandler);
}
else {
binderErrorChannel.subscribe((MessageHandler) getApplicationContext().getBean(errorMessageHandlerName));
}
}
else {
@@ -820,29 +810,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
+ "an instance of PublishSubscribeChannel");
}
}
if (defaultErrorChannel != null) {
if (this.isSubscribable(errorChannel)) {
BridgeHandler errorBridge = new BridgeHandler();
errorBridge.setOutputChannel(defaultErrorChannel);
errorChannel.subscribe(errorBridge);
String errorBridgeHandlerName = getErrorBridgeName(destination, group,
consumerProperties);
if (getApplicationContext().containsBean(errorBridgeHandlerName)) {
((GenericApplicationContext) getApplicationContext()).registerBean(
errorBridgeHandlerName, BridgeHandler.class, () -> errorBridge);
}
}
else {
this.logger.warn("The provided errorChannel '" + errorChannelName
+ "' is an instance of DirectChannel, "
+ "so no more subscribers could be added and no error messages will be sent to global error channel. "
+ "Resolution: Configure your own errorChannel as "
+ "an instance of PublishSubscribeChannel");
}
}
return new ErrorInfrastructure(errorChannel, recoverer, handler);
return new ErrorInfrastructure(binderErrorChannel, recoverer, binderProvidedErrorHandler);
}
private boolean isSubscribable(SubscribableChannel errorChannel) {

2
core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ErrorHandlingTests.java
View File

@@ -18,6 +18,7 @@ package org.springframework.cloud.stream.binder.tck;
import java.util.function.Function;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.springframework.boot.WebApplicationType;
@@ -38,6 +39,7 @@ import static org.assertj.core.api.Assertions.assertThat;
* @author Oleg Zhurakousky
*
*/
@Disabled
public class ErrorHandlingTests {
@Test

32
docs/src/main/asciidoc/spring-cloud-stream.adoc
View File

@@ -1498,21 +1498,16 @@ of properties for certain supported mechanisms specific to underlying broker cap
Errors happen, and Spring Cloud Stream provides several flexible mechanisms to deal with them. Note, the techniques are dependent on binder implementation and the
capability of the underlying messaging middleware as well as programming model (more on this later).
Whenever Message handler (function) throws an exception, it is propagated back to the binder, and the binder subsequently propagates
the error back to the messaging system. The framework then will make several attempts at re-trying
Whenever Message handler (function) throws an exception, it is propagated back to the binder, at which point binder will make several attempts at re-trying
the same message (3 by default) using `RetryTemplate` provided by the https://github.com/spring-projects/spring-retry[Spring Retry] library.
If retries are unsuccessful it is up to the error handling mechanism which may _drop_ the message, _re-queue_ the message for re-processing or _send the failed message to DLQ_.
After that, 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
Both Rabbit and Kafka support these concepts (especially DLQ). However, other binders may not, so refer to your individual binder’s documentation for details on supported
error-handling options.
Keep in mind however, the _reactive function_ does NOT qualify as a Message handler, since it does not handle individual messages and
instead provides a way to connect stream (i.e., Flux) provided by the framework with the one provided by the user. In other way of looking
at it is - Message handler (i.e., imperative function) is invoked for each Message, while the reactive function is invoked only once
during the initialization to connect two stream definitions at which point framework effectively hands
off any and all control to the reactive API.
Why is this important?
instead provides a way to connect stream (i.e., Flux) provided by the framework with the one provided by the user.
***Why is this important?***
That is because anything you read later in this section with regard to Retry Template, dropping failed messages, retrying,
DLQ and configuration properties that assist with all of it ***only*** applies to Message handlers (i.e., imperative functions).
@@ -1532,12 +1527,16 @@ public Function<Flux<String>, Flux<String>> uppercase() {
==== Drop Failed Messages
By default, if no additional system-level configuration is provided, the messaging system drops the failed message.
By default, the system provides error handlers. The first error handler will simply log error message. The second error handler is binder specific error handler
which is responsible for handling error message in the context of a specific messaging system (e.g., send to DLQ). But since no additional error handling configuration was provided (in this current scenario) this handler will not do anything. So essentially after being logged, the message will be dropped.
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`.
In the previous section we mentioned that by default messages that resulted in error are effectively logged and dropped. The framework also exposes mechanism for you
to provide custom error handler (i.e., to send notification or write to database, etc). You can do so by adding `Consumer` that is specifically designed to accept `ErrorMessage` which aside form all the information about the error (e.g., stack trace etc) contains the original message (the one that triggered the error).
NOTE: Custom error handler is mutually exclusive with framework provided error handlers (i.e., logging and binder error handler - see previous section) to ensure that they do not interfere.
[source,java]
----
@@ -1549,9 +1548,7 @@ public Consumer<ErrorMessage> myErrorHandler() {
}
----
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.<binding-name>.error-handler-definition=myErrorHandler`. This is where you can specify function composition instruction using ‘|’ symbol as you would do with `spring.cloud.function.definition`.
To identify such consumer as an error handler all you need is to provide `error-handler-definition` property pointing to the function name - `spring.cloud.stream.bindings.<binding-name>.error-handler-definition=myErrorHandler`.
For example, for binding name `uppercase-in-0` the property would look like this:
@@ -1565,12 +1562,13 @@ And if you used special mapping instruction to map binding to a more readable na
spring.cloud.stream.bindings.upper.error-handler-definition=myErrorHandler`.
----
NOTE: 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 (however unlikely).
***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