Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2735: Add errorChannel to ScatterGatherHandler

Browse Source 
Fixes spring-projects/spring-integration#2735

* Fix typos in Scatter-Gather JavaDocs
* Add Scatter-Gather error handling documentation

Doc polishing
This commit is contained in:
Artem Bilan
2019-02-05 15:25:04 -05:00
committed by Gary Russell
parent 5c46efe067
commit e8aa8618df
5 changed files with 189 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java
View File

@@ -86,6 +86,7 @@ import org.springframework.integration.transformer.MessageTransformingHandler;
import org.springframework.integration.transformer.MethodInvokingTransformer;
import org.springframework.integration.transformer.Transformer;
import org.springframework.integration.util.ClassUtils;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
@@ -2847,7 +2848,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* Can be {@code null}.
* @return the current {@link IntegrationFlowDefinition}.
*/
public B scatterGather(Consumer<RecipientListRouterSpec> scatterer, Consumer<AggregatorSpec> gatherer) {
public B scatterGather(Consumer<RecipientListRouterSpec> scatterer, @Nullable Consumer<AggregatorSpec> gatherer) {
return scatterGather(scatterer, gatherer, null);
}
@@ -2861,8 +2862,9 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* {@link ScatterGatherHandler} and its endpoint. Can be {@code null}.
* @return the current {@link IntegrationFlowDefinition}.
*/
public B scatterGather(Consumer<RecipientListRouterSpec> scatterer, Consumer<AggregatorSpec> gatherer,
Consumer<ScatterGatherSpec> scatterGather) {
public B scatterGather(Consumer<RecipientListRouterSpec> scatterer, @Nullable Consumer<AggregatorSpec> gatherer,
@Nullable Consumer<ScatterGatherSpec> scatterGather) {
Assert.notNull(scatterer, "'scatterer' must not be null");
RecipientListRouterSpec recipientListRouterSpec = new RecipientListRouterSpec();
scatterer.accept(recipientListRouterSpec);

15
spring-integration-core/src/main/java/org/springframework/integration/dsl/ScatterGatherSpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,6 +16,7 @@
package org.springframework.integration.dsl;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.scattergather.ScatterGatherHandler;
import org.springframework.messaging.MessageChannel;
@@ -58,4 +59,16 @@ public class ScatterGatherSpec extends ConsumerEndpointSpec<ScatterGatherSpec, S
return this;
}
/**
* Specify a {@link MessageChannel} bean name for async error processing.
* Defaults to {@link IntegrationContextUtils#ERROR_CHANNEL_BEAN_NAME}.
* @param errorChannel the {@link MessageChannel} bean name for async error processing.
* @return the current {@link ScatterGatherSpec} instance.
* @since 5.1.3
*/
public ScatterGatherSpec errorChannel(String errorChannel) {
this.handler.setErrorChannelName(errorChannel);
return this;
}
}

102
spring-integration-core/src/main/java/org/springframework/integration/scattergather/ScatterGatherHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2018 the original author or authors.
* Copyright 2014-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -17,14 +17,18 @@
package org.springframework.integration.scattergather;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.FixedSubscriberChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.endpoint.ReactiveStreamsConsumer;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.support.channel.HeaderChannelRegistry;
import org.springframework.messaging.Message;
@@ -32,7 +36,6 @@ import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.util.Assert;
@@ -57,6 +60,8 @@ public class ScatterGatherHandler extends AbstractReplyProducingMessageHandler i
private MessageChannel gatherChannel;
private String errorChannelName = IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME;
private long gatherTimeout = -1;
private AbstractEndpoint gatherEndpoint;
@@ -64,6 +69,13 @@ public class ScatterGatherHandler extends AbstractReplyProducingMessageHandler i
private HeaderChannelRegistry replyChannelRegistry;
public ScatterGatherHandler(MessageHandler scatterer, MessageHandler gatherer) {
this(new FixedSubscriberChannel(scatterer), gatherer);
Assert.notNull(scatterer, "'scatterer' must not be null");
Class<?> scattererClass = AopUtils.getTargetClass(scatterer);
checkClass(scattererClass, "org.springframework.integration.router.RecipientListRouter", "scatterer");
}
public ScatterGatherHandler(MessageChannel scatterChannel, MessageHandler gatherer) {
Assert.notNull(scatterChannel, "'scatterChannel' must not be null");
Assert.notNull(gatherer, "'gatherer' must not be null");
@@ -73,13 +85,6 @@ public class ScatterGatherHandler extends AbstractReplyProducingMessageHandler i
this.gatherer = gatherer;
}
public ScatterGatherHandler(MessageHandler scatterer, MessageHandler gatherer) {
this(new FixedSubscriberChannel(scatterer), gatherer);
Assert.notNull(scatterer, "'scatterer' must not be null");
Class<?> scattererClass = AopUtils.getTargetClass(scatterer);
checkClass(scattererClass, "org.springframework.integration.router.RecipientListRouter", "scatterer");
}
public void setGatherChannel(MessageChannel gatherChannel) {
this.gatherChannel = gatherChannel;
}
@@ -88,8 +93,20 @@ public class ScatterGatherHandler extends AbstractReplyProducingMessageHandler i
this.gatherTimeout = gatherTimeout;
}
/**
* Specify a {@link MessageChannel} bean name for async error processing.
* Defaults to {@link IntegrationContextUtils#ERROR_CHANNEL_BEAN_NAME}.
* @param errorChannelName the {@link MessageChannel} bean name for async error processing.
* @since 5.1.3
*/
public void setErrorChannelName(String errorChannelName) {
Assert.hasText(errorChannelName, "'errorChannelName' must not be empty.");
this.errorChannelName = errorChannelName;
}
@Override
protected void doInit() {
BeanFactory beanFactory = getBeanFactory();
if (this.gatherChannel == null) {
this.gatherChannel = new FixedSubscriberChannel(this.gatherer);
}
@@ -101,33 +118,39 @@ public class ScatterGatherHandler extends AbstractReplyProducingMessageHandler i
this.gatherEndpoint = new PollingConsumer((PollableChannel) this.gatherChannel, this.gatherer);
((PollingConsumer) this.gatherEndpoint).setReceiveTimeout(this.gatherTimeout);
}
else {
throw new MessagingException("Unsupported 'replyChannel' type [" + this.gatherChannel.getClass() + "]."
+ "SubscribableChannel or PollableChannel type are supported.");
else if (this.gatherChannel instanceof ReactiveStreamsSubscribableChannel) {
this.gatherEndpoint = new ReactiveStreamsConsumer(this.gatherChannel, this.gatherer);
}
this.gatherEndpoint.setBeanFactory(this.getBeanFactory());
else {
throw new BeanInitializationException("Unsupported 'replyChannel' type '" +
this.gatherChannel.getClass() + "'. " +
"'SubscribableChannel', 'PollableChannel' or 'ReactiveStreamsSubscribableChannel' " +
"types are supported.");
}
this.gatherEndpoint.setBeanFactory(beanFactory);
this.gatherEndpoint.afterPropertiesSet();
}
((MessageProducer) this.gatherer).setOutputChannel(new FixedSubscriberChannel(message -> {
MessageHeaders headers = message.getHeaders();
if (headers.containsKey(GATHER_RESULT_CHANNEL)) {
Object gatherResultChannel = headers.get(GATHER_RESULT_CHANNEL);
if (gatherResultChannel instanceof MessageChannel) {
messagingTemplate.send((MessageChannel) gatherResultChannel, message);
}
else if (gatherResultChannel instanceof String) {
messagingTemplate.send((String) gatherResultChannel, message);
}
}
else {
throw new MessageDeliveryException(message,
"The 'gatherResultChannel' header is required to delivery gather result.");
}
}));
((MessageProducer) this.gatherer)
.setOutputChannel(new FixedSubscriberChannel(message -> {
MessageHeaders headers = message.getHeaders();
if (headers.containsKey(GATHER_RESULT_CHANNEL)) {
Object gatherResultChannel = headers.get(GATHER_RESULT_CHANNEL);
if (gatherResultChannel instanceof MessageChannel) {
messagingTemplate.send((MessageChannel) gatherResultChannel, message);
}
else if (gatherResultChannel instanceof String) {
messagingTemplate.send((String) gatherResultChannel, message);
}
}
else {
throw new MessageDeliveryException(message,
"The 'gatherResultChannel' header is required to delivery gather result.");
}
}));
this.replyChannelRegistry = getBeanFactory()
.getBean(IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME,
this.replyChannelRegistry =
beanFactory.getBean(IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME,
HeaderChannelRegistry.class);
}
@@ -137,11 +160,13 @@ public class ScatterGatherHandler extends AbstractReplyProducingMessageHandler i
Object gatherResultChannelName = this.replyChannelRegistry.channelToChannelName(gatherResultChannel);
Message<?> scatterMessage = getMessageBuilderFactory()
.fromMessage(requestMessage)
.setHeader(GATHER_RESULT_CHANNEL, gatherResultChannelName)
.setReplyChannel(this.gatherChannel)
.build();
Message<?> scatterMessage =
getMessageBuilderFactory()
.fromMessage(requestMessage)
.setHeader(GATHER_RESULT_CHANNEL, gatherResultChannelName)
.setReplyChannel(this.gatherChannel)
.setErrorChannelName(this.errorChannelName)
.build();
this.messagingTemplate.send(this.scatterChannel, scatterMessage);
@@ -151,7 +176,7 @@ public class ScatterGatherHandler extends AbstractReplyProducingMessageHandler i
.fromMessage(gatherResult)
.removeHeader(GATHER_RESULT_CHANNEL)
.setHeader(MessageHeaders.REPLY_CHANNEL, requestMessage.getHeaders().getReplyChannel())
.build();
.setHeader(MessageHeaders.ERROR_CHANNEL, requestMessage.getHeaders().getErrorChannel());
}
return null;
@@ -179,7 +204,8 @@ public class ScatterGatherHandler extends AbstractReplyProducingMessageHandler i
private void checkClass(Class<?> gathererClass, String className, String type) throws LinkageError {
try {
Class<?> clazz = ClassUtils.forName(className, ClassUtils.getDefaultClassLoader());
Assert.isAssignable(clazz, gathererClass, "the '" + type + "' must be an " + className + " instance");
Assert.isAssignable(clazz, gathererClass, () -> "the '" + type + "' must be an " + className + " " +
"instance");
}
catch (ClassNotFoundException e) {
throw new IllegalStateException("The class for '" + className + "' cannot be loaded", e);

57
spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/RouterTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2018 the original author or authors.
* Copyright 2016-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -39,8 +39,10 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.annotation.Router;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.config.EnableMessageHistory;
@@ -550,8 +552,9 @@ public class RouterTests {
Object payload = bestQuoteMessage.getPayload();
assertThat(payload, instanceOf(String.class));
List<?> topSequenceDetails =
(List<?>) bestQuoteMessage.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS, List.class)
.get(0);
(List<?>) bestQuoteMessage.getHeaders()
.get(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS, List.class)
.get(0);
assertEquals(request.getHeaders().getId(),
bestQuoteMessage.getHeaders().get(IntegrationMessageHeaderAccessor.CORRELATION_ID));
@@ -566,6 +569,26 @@ public class RouterTests {
topSequenceDetails.get(2));
}
@Autowired
@Qualifier("scatterGatherAndExecutorChannelSubFlow.input")
private MessageChannel scatterGatherAndExecutorChannelSubFlowInput;
@Test
public void testScatterGatherWithExecutorChannelSubFlow() {
QueueChannel replyChannel = new QueueChannel();
Message<?> testMessage =
MessageBuilder.withPayload("test")
.setReplyChannel(replyChannel)
.build();
this.scatterGatherAndExecutorChannelSubFlowInput.send(testMessage);
Message<?> receive = replyChannel.receive(10_000);
assertNotNull(receive);
Object payload = receive.getPayload();
assertThat(payload, instanceOf(List.class));
assertThat(((List) payload).get(1), instanceOf(RuntimeException.class));
}
@Configuration
@EnableIntegration
@@ -689,7 +712,9 @@ public class RouterTests {
@Bean
public IntegrationFlow routeMultiMethodInvocationFlow() {
return IntegrationFlows.from("routerMultiInput")
.route(String.class, p -> p.equals("foo") || p.equals("bar") ? new String[] { "foo", "bar" } : null,
.route(String.class, p -> p.equals("foo") || p.equals("bar")
? new String[] { "foo", "bar" }
: null,
s -> s.suffix("-channel"))
.get();
}
@@ -832,6 +857,30 @@ public class RouterTests {
.collect(Collectors.joining("\n")));
}
@Bean
public IntegrationFlow scatterGatherAndExecutorChannelSubFlow(TaskExecutor taskExecutor) {
return f -> f
.scatterGather(
scatterer -> scatterer
.applySequence(true)
.recipientFlow(f1 -> f1.transform(p -> "Sub-flow#1"))
.recipientFlow(f2 -> f2
.channel(c -> c.executor(taskExecutor))
.transform(p -> {
throw new RuntimeException("Sub-flow#2");
})),
null,
s -> s.errorChannel("scatterGatherErrorChannel"));
}
@ServiceActivator(inputChannel = "scatterGatherErrorChannel")
public Message<?> processAsyncScatterError(MessagingException payload) {
return MessageBuilder.withPayload(payload.getCause().getCause())
.copyHeaders(payload.getFailedMessage().getHeaders())
.build();
}
}
private static class RoutingTestBean {

53
src/reference/asciidoc/scatter-gather.adoc
View File

@@ -153,3 +153,56 @@ Mutually exclusive with `scatter-channel` attribute.
<13> The `<aggregator>` options.
Required.
====
[[scatter-gather-error-handling]]
==== Error Handling
Since Scatter-Gather is a multi request-reply component, error handling has some extra complexity.
In some cases, it is better to just catch and ignore downstream exceptions if the `ReleaseStrategy` allows the process to finish with fewer replies than requests.
In other cases something like a "`compensation message`" should be considered for returning from sub-flow, when an error happens.
Every async sub-flow should be configured with a `errorChannel` header for the proper error message sending from the `MessagePublishingErrorHandler`.
Otherwise, an error will be sent to the global `errorChannel` with the common error handling logic.
See <<namespace-errorhandler>> for more information about async error processing.
Synchronous flows may use an `ExpressionEvaluatingRequestHandlerAdvice` for ignoring the exception or returning a compensation message.
When an exception is thrown from one of the sub-flows to the `ScatterGatherHandler`, it is just re-thrown to upstream.
This way all other sub-flows will work for nothing and their replies are going to be ignored in the `ScatterGatherHandler`.
This might be an expected behavior sometimes, but in most cases it would be better to handle the error in the particular sub-flow without impacting all others and the expectations in the gatherer.
Starting with version 5.1.3, the `ScatterGatherHandler` is supplied with the `errorChannelName` option.
It is populated to the `errorChannel` header of the scatter message and is used in the when async error happens or can be used in the regular synchronous sub-flow for directly sending an error message.
The sample configuration below demonstrates async error handling by returning a compensation message:
====
[source,java]
----
@Bean
public IntegrationFlow scatterGatherAndExecutorChannelSubFlow(TaskExecutor taskExecutor) {
return f -> f
.scatterGather(
scatterer -> scatterer
.applySequence(true)
.recipientFlow(f1 -> f1.transform(p -> "Sub-flow#1"))
.recipientFlow(f2 -> f2
.channel(c -> c.executor(taskExecutor))
.transform(p -> {
throw new RuntimeException("Sub-flow#2");
})),
null,
s -> s.errorChannel("scatterGatherErrorChannel"));
}
@ServiceActivator(inputChannel = "scatterGatherErrorChannel")
public Message<?> processAsyncScatterError(MessagingException payload) {
return MessageBuilder.withPayload(payload.getCause().getCause())
.copyHeaders(payload.getFailedMessage().getHeaders())
.build();
}
----
====
To produce a proper reply, we have to copy headers (including `replyChannel` and `errorChannel`) from the `failedMessage` of the `MessagingException` that has been sent to the `scatterGatherErrorChannel` by the `MessagePublishingErrorHandler`.
This way the target exception is returned to the gatherer of the `ScatterGatherHandler` for reply messages group completion.
Such an exception `payload` can be filtered out in the `MessageGroupProcessor` of the gatherer or processed other way downstream, after the scatter-gather endpoint.