Skip to content

GH-3189: Properly handle async in ScatterGatherHandler #10469

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Oct 8, 2025
+63 −20
Merged

GH-3189: Properly handle async in ScatterGatherHandler #10469

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
09f382a
GH-3189: Properly handle `async` in `ScatterGatherHandler`
artembilan Oct 7, 2025
13c0903
Mention `Mono` reply in the `ScatterGatherHandler` Javadocs
artembilan Oct 8, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 43 additions & 13 deletions ...ore/src/main/java/org/springframework/integration/scattergather/ScatterGatherHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,11 @@

package org.springframework.integration.scattergather;

import java.time.Duration;

import org.jspecify.annotations.Nullable;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;

import org.springframework.aop.support.AopUtils;
import org.springframework.beans.factory.BeanFactory;
Expand All @@ -32,6 +36,7 @@
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.endpoint.ReactiveStreamsConsumer;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.integration.support.management.ManageableLifecycle;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
Expand All @@ -48,6 +53,9 @@
/**
* The {@link MessageHandler} implementation for the
* <a href="https://www.enterpriseintegrationpatterns.com/BroadcastAggregate.html">Scatter-Gather</a> EIP pattern.
* <p>
* When {@link #setAsync(boolean)} is {@code true}, the {@link ScatterGatherHandler} produces
* a {@link Mono} as a reply based on the gather result.
*
* @author Artem Bilan
* @author Abdul Zaheer
Expand Down Expand Up @@ -146,11 +154,11 @@ public Message<?> preSend(Message<?> message, MessageChannel channel) {
}

});
if (this.gatherChannel instanceof SubscribableChannel) {
this.gatherEndpoint = new EventDrivenConsumer((SubscribableChannel) this.gatherChannel, this.gatherer);
if (this.gatherChannel instanceof SubscribableChannel subscribableChannel) {
this.gatherEndpoint = new EventDrivenConsumer(subscribableChannel, this.gatherer);
}
else if (this.gatherChannel instanceof PollableChannel) {
this.gatherEndpoint = new PollingConsumer((PollableChannel) this.gatherChannel, this.gatherer);
else if (this.gatherChannel instanceof PollableChannel pollableChannel) {
this.gatherEndpoint = new PollingConsumer(pollableChannel, this.gatherer);
((PollingConsumer) this.gatherEndpoint).setReceiveTimeout(this.gatherTimeout);
}
else if (this.gatherChannel instanceof ReactiveStreamsSubscribableChannel) {
Expand Down Expand Up @@ -191,7 +199,18 @@ private Message<?> enhanceScatterReplyMessage(Message<?> message) {
@Override
protected @Nullable Object handleRequestMessage(Message<?> requestMessage) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we need to enumerate that a mono is returned when async is true else a AbstractIntegrationMessageBuilder is returned?

MessageHeaders requestMessageHeaders = requestMessage.getHeaders();
PollableChannel gatherResultChannel = new QueueChannel();
boolean async = isAsync();
MessageChannel gatherResultChannel;
Sinks.One<Message<?>> replyMono;

if (async) {
replyMono = Sinks.one();
gatherResultChannel = (message, timeout) -> replyMono.tryEmitValue(message).isSuccess();
}
else {
replyMono = null;
gatherResultChannel = new QueueChannel();
}

Message<?> scatterMessage =
getMessageBuilderFactory()
Expand All @@ -204,17 +223,28 @@ private Message<?> enhanceScatterReplyMessage(Message<?> message) {

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

Message<?> gatherResult = gatherResultChannel.receive(this.gatherTimeout);
if (gatherResult != null) {
return getMessageBuilderFactory()
.fromMessage(gatherResult)
.removeHeaders(GATHER_RESULT_CHANNEL, ORIGINAL_ERROR_CHANNEL,
MessageHeaders.REPLY_CHANNEL, MessageHeaders.ERROR_CHANNEL);
if (replyMono != null) {
return replyMono.asMono()
.map(this::replyFromGatherResult)
.timeout(Duration.ofMillis(this.gatherTimeout), Mono.empty());
}
else {
Message<?> gatherResult = ((PollableChannel) gatherResultChannel).receive(this.gatherTimeout);
if (gatherResult != null) {
return replyFromGatherResult(gatherResult);
}
}

return null;
}

private AbstractIntegrationMessageBuilder<?> replyFromGatherResult(Message<?> gatherResult) {
return getMessageBuilderFactory()
.fromMessage(gatherResult)
.removeHeaders(GATHER_RESULT_CHANNEL, ORIGINAL_ERROR_CHANNEL,
MessageHeaders.REPLY_CHANNEL, MessageHeaders.ERROR_CHANNEL);
}

@Override
public void start() {
if (this.gatherEndpoint != null) {
Expand All @@ -240,8 +270,8 @@ private static void checkClass(Class<?> gathererClass, String className, String
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);
catch (ClassNotFoundException ex) {
throw new IllegalStateException("The class for '" + className + "' cannot be loaded", ex);
}
}

Expand Down
23 changes: 16 additions & 7 deletions ...tegration-core/src/test/java/org/springframework/integration/dsl/routers/RouterTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

package org.springframework.integration.dsl.routers;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
Expand Down Expand Up @@ -463,15 +464,21 @@ public void testRouterAsNonLastComponent() {
@Test
public void testScatterGather() {
QueueChannel replyChannel = new QueueChannel();
Message<String> request = MessageBuilder.withPayload("foo")
Message<String> request = MessageBuilder.withPayload("test")
.setReplyChannel(replyChannel)
.build();
this.scatterGatherFlowInput.send(request);
Message<?> bestQuoteMessage = replyChannel.receive(10000);
assertThat(bestQuoteMessage).isNotNull();
Object payload = bestQuoteMessage.getPayload();
assertThat(payload).isInstanceOf(List.class);
assertThat(((List<?>) payload).size()).isGreaterThanOrEqualTo(1);
assertThat(bestQuoteMessage)
.extracting(Message::getPayload)
.asInstanceOf(InstanceOfAssertFactories.LIST)
.hasSizeGreaterThanOrEqualTo(1)
.first()
.asInstanceOf(InstanceOfAssertFactories.type(Message.class))
.extracting(Message::getHeaders)
.asInstanceOf(InstanceOfAssertFactories.MAP)
.extractingByKey("gatherResultChannel")
.isNotInstanceOf(PollableChannel.class);
}

@Autowired
Expand Down Expand Up @@ -859,9 +866,11 @@ public IntegrationFlow scatterGatherFlow() {
group.size() == 3 ||
group.getMessages()
.stream()
.anyMatch(m -> (Double) m.getPayload() > 5)),
.anyMatch(m -> (Double) m.getPayload() > 5))
.outputProcessor(group -> new ArrayList<>(group.getMessages())),
scatterGather -> scatterGather
.gatherTimeout(10_000));
.gatherTimeout(10_000)
.async(true));
}

@Bean
Expand Down
4 changes: 4 additions & 0 deletions src/reference/antora/modules/ROOT/pages/scatter-gather.adoc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -154,6 +154,10 @@ Mutually exclusive with `scatter-channel` attribute.
<13> The `<aggregator>` options.
Required.

NOTE: Starting with version `6.5.3`, when a `ScatterGatherHandler` is configured for the `async = true` option, the request message handling thread is not blocked anymore waiting for a gather result on an internal `((PollableChannel) gatherResultChannel).receive(this.gatherTimeout)` operation.
Instead, a `reactor.core.publisher.Mono` is returned as a reply object based on a gather result eventually produced from the `gatherResultChannel`.
Such a `Mono` is handled then according to the xref:reactive-streams.adoc#reactive-reply-payload[Reactive Streams support] in the framework.

[[scatter-gather-error-handling]]
== Error Handling

Expand Down