shashank created CAMEL-24959:
--------------------------------
Summary: Split with streaming and parallelProcessing completes
before its parts and loses the aggregated result when the iterator returns null
after hasNext()=true
Key: CAMEL-24959
URL: https://issues.apache.org/jira/browse/CAMEL-24959
Project: Camel
Issue Type: Bug
Components: camel-core
Reporter: shashank
An iterator may answer {{hasNext()}} with true and then return null from
{{next()}}. The Splitter skips such null parts (CAMEL-9745,
{{SplitIteratorNullTest}}). With {{streaming()}} + {{parallelProcessing()}}, a
trailing null part makes the split complete at once, while the parts it already
sent are still running. The aggregated result is lost and the route continues
with the original message.
In parallel mode {{MulticastReactiveTask.run()}} re-schedules itself whenever
{{hasNext}} was true. The next run skips the null in
{{getNextProcessorExchangePair()}}, finds no pair and calls
{{doDone(result.get(), true)}} immediately. The replies of the in-flight parts
are then ignored by the {{done}} check in {{aggregate()}}. Sequential mode is
not affected, because there the next run starts only after the previous part
completed.
Reproduced against 4.23.0-SNAPSHOT with the iterator from
{{SplitIteratorNullTest}} (A, B, C, then {{hasNext()=true}} / {{next()=null}}),
a concatenating strategy and a 300 ms sub-route:
{noformat}
split(...).streaming() -> body=ABC after 940 ms
split(...).parallelProcessing() -> body=ABC after 308 ms
split(...).streaming().parallelProcessing() -> body=<original> after 3 ms; A,
B, C processed afterwards and never aggregated
{noformat}
CAMEL-21114 reported the same symptom for transacted routes with the zip
splitter (fixed there by changing the transacted scheduling). The reactive
parallel path still has it.
Proposed fix: when the task finds no more pairs, treat it like "last pair sent"
(new {{MulticastTask.doDoneNoMorePairs()}}, called from
{{MulticastReactiveTask.run()}} instead of {{doDone}}). Under the task lock it
sets {{allSent}}, aggregates the completed exchanges that are queued, and calls
{{doDone}} only if everything sent has been aggregated. Otherwise
{{aggregate()}} completes when it aggregates the last part, as it already does
when the last pair is known to be the last one.
* The lock is needed. A lock-free "set {{allSent}}, then compare
{{nbAggregated}} with {{nbExchangeSent}}" could see the counter the timeout is
incrementing for the missing indices and complete as a normal completion while
{{timeout()}} is still calling the strategy (a TLA+ model of the task finds
this interleaving).
* Aggregating under the lock is needed too. {{aggregate()}} only does
{{tryLock()}}, so a part that completes while the task holds the lock gives up
and relies on the lock holder to poll its result.
* The poll loop of {{aggregate()}} moved unchanged into a private
{{aggregateCompleted()}}, which both methods use.
* Sequential mode takes the same path. There the previous part is always
aggregated before the next run, so it behaves as before: complete at once.
* The transacted task is unchanged (CAMEL-21114 fixed it there).
A PR with the fix follows, with regression test
{{SplitParallelStreamingIteratorNullTest}}. It uses the iterator from
{{SplitIteratorNullTest}}, and the parts wait until the iterator's final
{{hasNext()}} returned false, so they are still in flight when the splitter
finds no more parts. Test methods: {{testSplitStreamingParallel}} (the bug) and
{{testSplitStreaming}} (sequential control). Without the fix,
{{testSplitStreamingParallel}} fails with {{The split should return the
aggregated parts, but returned:
org.apache.camel.processor.SplitParallelStreamingIteratorNullTest$MyIterator@70cc4471
==> expected: <ABC> ...}} (the original body).
Found with a TLA+ model of the multicast task, then reproduced against
4.23.0-SNAPSHOT.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
