From e8aa8618dfbaa155419500df9d1d374a1da1f9a7 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 5 Feb 2019 15:25:04 -0500 Subject: [PATCH] GH-2735: Add errorChannel to ScatterGatherHandler Fixes spring-projects/spring-integration#2735 * Fix typos in Scatter-Gather JavaDocs * Add Scatter-Gather error handling documentation Doc polishing --- .../dsl/IntegrationFlowDefinition.java | 8 +- .../integration/dsl/ScatterGatherSpec.java | 15 ++- .../scattergather/ScatterGatherHandler.java | 102 +++++++++++------- .../integration/dsl/routers/RouterTests.java | 57 +++++++++- src/reference/asciidoc/scatter-gather.adoc | 53 +++++++++ 5 files changed, 189 insertions(+), 46 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java index 45639ed328..a694b43998 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java @@ -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 scatterer, Consumer gatherer) { + public B scatterGather(Consumer scatterer, @Nullable Consumer gatherer) { return scatterGather(scatterer, gatherer, null); } @@ -2861,8 +2862,9 @@ public abstract class IntegrationFlowDefinition scatterer, Consumer gatherer, - Consumer scatterGather) { + public B scatterGather(Consumer scatterer, @Nullable Consumer gatherer, + @Nullable Consumer scatterGather) { + Assert.notNull(scatterer, "'scatterer' must not be null"); RecipientListRouterSpec recipientListRouterSpec = new RecipientListRouterSpec(); scatterer.accept(recipientListRouterSpec); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/ScatterGatherSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/ScatterGatherSpec.java index f22f3f0994..c0398f48af 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/ScatterGatherSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/ScatterGatherSpec.java @@ -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 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); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/RouterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/RouterTests.java index 2c290d3c79..ac58a07ed0 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/RouterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/RouterTests.java @@ -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 { diff --git a/src/reference/asciidoc/scatter-gather.adoc b/src/reference/asciidoc/scatter-gather.adoc index c593e1a3fb..f00fa296af 100644 --- a/src/reference/asciidoc/scatter-gather.adoc +++ b/src/reference/asciidoc/scatter-gather.adoc @@ -153,3 +153,56 @@ Mutually exclusive with `scatter-channel` attribute. <13> The `` 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 <> 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.