Spring-Integration 5.1.2/5.1.3(pre-release Build): Issue with Scatter Gather Error in Gatherer not sent back to requestor

[sirimamilla]

Affects Version(s): \5.1.2/5.1.3(SNAPSHOT BUILD)

---

Hi @garyrussell, @artembilan , @dturanski ,

Good Morning,

We are facing a new issue with Spring Integration 5.1.2 & 5.1.3 Snapshot Build where in any error occurs in the Aggregtor when OutputProcessor is used, the error is not propagated to the caller. the caller thread is hung indefinitely.

Please find the attached Github source replicating the issue. Please run the ScatterGatherTestAggregatorError test to replicate the issue.

Source: https://github.com/sirimamilla/ScatterGatherTest

Thanks,
Jayadev

[artembilan]

So, the situation is still with an async hand off with an ExecutorChannel in some-subflow.
The point is that an Aggregator (gatherer in case of Scatter-Gather) produces result (by default) really on the thread of the message which has initiated a group completion.
In our case that is a thread from the mentioned ExecutorChannel. Therefore any errors happened in the aggregation function is going to be sent to the MessagePublishingErrorHandler.
Which in your case is ended up with this:

  @ServiceActivator(inputChannel = "scatterGatherErrorChannel")
  public Message<?> processAsyncScatterError(MessagingException payload) {
    return MessageBuilder.withPayload(payload.getCause())
            .copyHeaders(payload.getFailedMessage().getHeaders())
            .build();
  }

Unfortunately, the replyChannel header in this case is just an internal gatherChannel populated by the ScatterGatherHandler, so we are going to have something like stack overflow pushing the message to the gatherer again and again.

This is something wrong from the ScatterGatherHandler and as a fix for this issue I suggest to reinstate original headers (replyChannel/errorChannel) from the request message. Just because we need a gatherChannel only for scatter phase to be able to pick up all the replies in a single place like gatherer.

As a workaround I suggest to have an explicit gatherChannel() configuration with its own local error handling to return a compensation message for the upstream awaiting gateway.

[sirimamilla]

Hi @artembilan ,

Even if I have a separate error channel for the gatherer, it will not help as the headers in the error message still has the scatterGatherErrorChannel as the error channel.

When Producing the Success reply, Gatherer replaces the reply & Error channel back to original reply and error channels from the request message received at Scatter Gather Channel. Similar behaviour is needed for the error flows as well.

Alternatively, Original request, processing of which caused error may also be wrapped in the Messagehandling Exception. So that in the error Handler, we can replace the errorChannel & replyChannel depending on whether to throw exception or Send Success Message.

This may solve the issue in both success & error scenarios.

Thanks,
Jayadev

[artembilan]

I don't have an out-of-the-box solution yet, but as a work around you can store an original message in the headers before sending to the Scatter-Gather. So, this way you indeed will have access to original reply and error channel headers anywhere in sub-flows or in the gatherer phase.

[artembilan]

OK. I found the solution with remapping original headers.
Will PR shortly...

[artembilan]

@sirimamilla ,

Please, take a look into the provided PR #2750 . Especially pay attention to the test config:

		@Bean
		public IntegrationFlow propagateErrorFromGatherer(TaskExecutor taskExecutor) {
			return IntegrationFlows.from(Function.class)
					.scatterGather(s -> s
									.applySequence(true)
									.recipientFlow(subFlow -> subFlow
											.channel(c -> c.executor(taskExecutor))
											.transform(p -> "foo")),
							g -> g
									.outputProcessor(group -> {
										throw new RuntimeException("intentional");
									}),
							sg -> sg.gatherTimeout(100))
					.get();
		}

So, with this fix your goal should be achieved.
Although it is important to keep in mind that gatherTimeout, which is "wait for reply forever" by default.