Scatter Gather is getting hung when Nested Scatter Gathers are used

[sirimamilla]

5.2.3.log
5.1.3.log

Affects Version(s): <Spring Integration version>
5.1.7 and above
5.2.3

Hi,

Scatter Gather is failing when nested scatter Gathers are used with latest versions of Spring Integration. we tried with 5.1.7, 5.1.9 and 5.2.3. Same is working fine in 5.1.6 and below.
Here is the sample project
https://github.com/sirimamilla/ScatterGatherTest

Test case ScatterGatherTestSuccess runs indefinitely in this project.

when analyzed, scatter gather is using origialReplyChannel and originalErrorChannel Headers to preserve the previous original reply and error channels.

Inner scatterGather is responding correctly to the outer scatter gather. But outer scatter gather is failing to respond as it is finding its own gatherchannel in original replychannel header and sending response to gather channel and waiting indefinitely to continue.

This issue is blocking our version upgrade from spring boot 2.1.3 to 2.2.4. Please assist.

Regards,
Jayadev

[sirimamilla]

Hi,

On Further analysis of issue, in version 5.1.3, reply channels are handled in handleRequestMessage. Reply channels are directly taken from the request message headers. Also no additional headers are added to the request sent to the subflows.

@Override
	protected Object handleRequestMessage(Message<?> requestMessage) {
		PollableChannel gatherResultChannel = new QueueChannel();

		Object gatherResultChannelName = this.replyChannelRegistry.channelToChannelName(gatherResultChannel);

		Message<?> scatterMessage = getMessageBuilderFactory()
				.fromMessage(requestMessage)
				.setHeader(GATHER_RESULT_CHANNEL, gatherResultChannelName)
				.setReplyChannel(this.gatherChannel)
				.build();

		this.messagingTemplate.send(this.scatterChannel, scatterMessage);

		Message<?> gatherResult = gatherResultChannel.receive(this.gatherTimeout);
		if (gatherResult != null) {
			return getMessageBuilderFactory()
					.fromMessage(gatherResult)
					.removeHeader(GATHER_RESULT_CHANNEL)
					.setHeader(MessageHeaders.REPLY_CHANNEL, requestMessage.getHeaders().getReplyChannel())
					.build();
		}

		return null;
	}

In 5.2.4 version, ReplyChannel and ErrorChannel are copied to headers ORIGINAL_REPLY_CHANNEL, ORIGINAL_ERROR_CHANNEL. This process is not taking care of any ORIGINAL_REPLY_CHANNEL, ORIGINAL_ERROR_CHANNEL already available in request headers. if these headers are already present in the headers, they will be overridden without being preserved for the reply path.

This is causing issue in the nested ScatterGather situations.

@Override
	protected Object handleRequestMessage(Message<?> requestMessage) {
		MessageHeaders requestMessageHeaders = requestMessage.getHeaders();
		PollableChannel gatherResultChannel = new QueueChannel();

		Message<?> scatterMessage =
				getMessageBuilderFactory()
						.fromMessage(requestMessage)
						.setHeader(GATHER_RESULT_CHANNEL, gatherResultChannel)
						.setHeader(ORIGINAL_REPLY_CHANNEL, requestMessageHeaders.getReplyChannel())
						.setHeader(ORIGINAL_ERROR_CHANNEL, requestMessageHeaders.getErrorChannel())
						.setReplyChannel(this.gatherChannel)
						.setErrorChannelName(this.errorChannelName)
						.build();

		this.messagingTemplate.send(this.scatterChannel, scatterMessage);

		return gatherResultChannel.receive(this.gatherTimeout);
	}

private Message<?> enhanceScatterReplyMessage(Message<?> message) {
		MessageHeaders headers = message.getHeaders();
		return getMessageBuilderFactory()
				.fromMessage(message)
				.setHeader(MessageHeaders.REPLY_CHANNEL, headers.get(ORIGINAL_REPLY_CHANNEL))
				.setHeader(MessageHeaders.ERROR_CHANNEL, headers.get(ORIGINAL_ERROR_CHANNEL))
				.removeHeaders(ORIGINAL_REPLY_CHANNEL, ORIGINAL_ERROR_CHANNEL)
				.build();
	}

[sirimamilla]

Hi,

I have done fix for the issue. But I am not able to raise a pull request as i dont have access to raise pull request. Please find the attached code to fix the issue with Nested channels.

There is further scope to refine the scatter gather.

ScatterGatherHandler.zip

[artembilan]

@sirimamilla ,

thank you very much for the report!

I'll take a look into this today.

I am not able to raise a pull request as i dont have access to raise pull request

I wonder if you have forked the project, created a branch for the fix and then you can try to Pull Request it.

See instructions here: https://github.com/spring-projects/spring-integration/blob/master/CONTRIBUTING.adoc

[artembilan]

I'm sorry, would you mind to be more specific.
I tried your projectwith all the versions down to Spring Integration 5.1.3 (5.1.2 doesn't have an errorChannel() on the ScatterGatherSpec yet) and all tests pass for me unless ScatterGatherTestAggregatorError().

So, what we are looking for?
Would you mind to minimize the project the way only about a problem and point, please, in which exactly version it works as expected.

Meanwhile I'll try to analyze your code...

[artembilan]

I have fixed your problem with this sg->sg.errorChannel("scatterGatherErrorChannel").gatherTimeout(1000). Pay attention to the gatherTimeout(1000).

The real problem that you block the main thread waiting for gatherer reply with the return gatherResultChannel.receive(this.gatherTimeout);.
At the same time the your gatherer (g->g.outputProcessor(this::processGroup)) doesn't finish its work but raises an exception which from its thread goes to the replyChannel of the gateway originating the request.
And since gateway is blocked already waiting for reply it doesn't go to receive of its reply channel.
Essentailly with such a thread shifting downstream you make a dead lock: you would like to let it to deal with an error, but that error doesn't release the main thread.

I'm not sure yet that can do anything on the matter, more over we have documented such a situation to make you ready: https://docs.spring.io/spring-integration/docs/5.2.3.RELEASE/reference/html/message-routing.html#scatter-gather-error-handling

finite gatherTimeout must be configured for the ScatterGatherHandler. Otherwise it is going to be blocked waiting for a reply from the gatherer forever, by default.

So, I'll wait for your feedback, but sounds like "Works as Designed"

[sirimamilla]

Hi Artem, it seems I didnt properly explain. Test case ScatterGatherTestSuccess always hangs. Not sure how that works for you.

GatherTumeout is to handle timeout scenarios.
In this testcase there is no channel timing out.

[artembilan]

Ok. Try to have only that one single test. Please, come back to me with a minimal reproducing sample and with consistent versions. In one comment you claimed one version, the you talk about different. And in your sample we have a third one.

Thanks for understanding!

[sirimamilla]

Hi @artembilan,

Good Morning.

I have updated the test project to have only failing test case(hangs forever) with version 5.2.3.
I have added gatherTimeout as per your suggestion above.

Same test case works perfectly fine with version 5.1.3

Thanks,
Jayadev

[artembilan]

Good. I’ll take a look in my morning in 12 hours 😀

[artembilan]

works perfectly fine with version 5.1.3

You still are confusing me. Does it work with version 5.1.6 because in the linked PR you claim that it stopped woriking since 5.1.7.

[sirimamilla]

It works with 5.1.4, 5.1.5, 5.1.6 as well

[artembilan]

OK. Thanks. BTW, you can override versions in your Spring Boot application using specific property. See more here: https://docs.spring.io/spring-boot/docs/2.2.4.RELEASE/reference/html/howto.html#howto-customize-dependency-versions

I mean you have overidden the version for spring-integration-core, but left it for spring-integration-test, which may cause some versions conflicts at runtime.