Auto-completion of the Message group and race condition between the Reaper and discard-channel [INT-2014]

[spring-operator]

Oleg Zhurakousky opened INT-2014 and commented

MessageGroup is considered complete if it has accumulated the amount of messages specified by the sequenceSize. However CorrelatingMessageHandler can still return 'false' on canRelease() call if custom release strategy is provided. For example, i may want to batch up messages in groups (as the use case in the forum) of 5 while my sequence size is 7. This means that the first batch of 5 messages will be released as soon as MessageGroup has 5 messages. The other two Messages will complete the MessageGroup and based on the current code in CorrelatingMessageHandler these two Messages will be sent to a discard channel unless Reaper gets to them first. So its basically one big race condition with unpredictable results.
This goes back to the mutual exclusivity. If send-partial-result-on-expiry is set to 'true', then CorrelatingMessageHandler should assume that Reaper has control over the MessageGroup and not send messages to the discardChannel

 else if (!sendPartialResultOnExpiry && group.isComplete())  instead of what we have now else if (group.isComplete())

---

Affects: 2.0.5

Issue Links:

• Remove the role of the Sequence Size for the aggregators that are not relying on the default SequenceSizeReleaseStrategy [INT-2134] #6116 Remove the role of the Sequence Size for the aggregators that are not relying on the default SequenceSizeReleaseStrategy

[spring-operator]

Mark Fisher commented

If there is a custom ReleaseStrategy, shouldn't that take full responsibility? In other words, the SequenceSizeReleaseStrategy is the default, but if you replace it, then the sequence size is no longer relevant unless you deal with it in some way within your own custom strategy.

[spring-operator]

Oleg Zhurakousky commented

Yes, but please read comment I left in the PR #75

[spring-operator]

Oleg Zhurakousky commented

After several discussions we've determined that in relation to this issue the Aggregator works as designed. The use case that is mentioned in the description really talks about "batching", and possibly unbounded (no sequence number) aggregator and could easily be handled today by a simple SpEL-based release strategy (e.g., size() == 4 if one wants to release messages in batches of 4). If such aggregation is bounded than it would be advisable to tune your release strategy to make sure that there is no possibility of a leftover messages (which will only be release based on the timeout if used in conjunction with the reaper), otherwise if it is still the case, one can implement a more complex and stateful ReleaseStrategy that maintains som type of counter and realizes that there wil be no more messages coming.