Scatter Gather with parallel flow (Timeout in aggregator)

Viewed 167

I've been trying to add a timeout in the gather to don't wait that every flow finished. but when I added the timeout doesn't work because the aggregator waits that each flow finished.

@Bean
public IntegrationFlow queueFlow(LogicService service) {    
        return f -> f.scatterGather(scatterer -> scatterer
                                .applySequence(true)
                                .recipientFlow(aFlow(service))
                                .recipientFlow(bFlow(service))
                        , aggregatorSpec -> aggregatorSpec.groupTimeout(2000L)) 

E.g of my flows one of them has 2 secs of delay and the other one 4 secs

public IntegrationFlow bFlow(LogicService service) {
        return IntegrationFlows.from(MessageChannels.executor(Executors.newCachedThreadPool()))
                .handle(service::callFakeServiceTimeout2)
                .transform((MessageDomain.class), message -> {
                    message.setMessage(message.getMessage().toUpperCase());
                    return message;
                }).get();
    } 

I use Executors.newCachedThreadPool() to run parallel. I'd like to release each message that was contained until the timeout is fulfilled

Another approach that I've been testing was to use a default gatherer and in scatterGather set the gatherTimeout but I don't know if I'm missing something Approach gatherTimeout

UPDATE

All the approaches given in the comments were tested and work normally, the only problem is that each action is evaluated over the message group creation. and the message group is created just until the first message arrived. The ideal approach is having an option of valid at the moment when the scatterer distributes the request message.

My temporal solution was to use a release strategy ad hoc applying a GroupConditionProvider which reads a custom header that I created when I send the message through the gateway. The only concern of this is that the release strategy only will be executed when arriving at a new message or I set a group time out.

1 Answers

The groupTimeout on the aggregator is not enough to release the group. If you don't get the whole group on that timeout, then it is going to be discarded. See sendPartialResultOnExpiry option: https://docs.spring.io/spring-integration/reference/html/message-routing.html#agg-and-group-to

If send-partial-result-on-expiry is true, existing messages in the (partial) MessageGroup are released as a normal aggregator reply message to the output-channel. Otherwise, it is discarded.

The gatherTimeout is good to have if you expect no replies from the gatherer at all. So, this way you won't block the scatter-gather thread forever: https://docs.spring.io/spring-integration/reference/html/message-routing.html#scatter-gather-error-handling

Related