[
https://issues.apache.org/jira/browse/CAMEL-24959?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen resolved CAMEL-24959.
---------------------------------
Fix Version/s: 4.23.0
Resolution: Fixed
The fix is merged on main, so it is in Camel 4.23.0:
* 4700e7f3c7aa CAMEL-24959: camel-core - Do not complete a parallel streaming
split before its parts when the iterator ends with a null
* 3e53ef0f81d3 CAMEL-24959: camel-core - Make the new test class package-private
* c96a268d7bdf CAMEL-24959: camel-core - Make doDoneNoMorePairs private
Resolving, as the ticket was not updated when the PR was merged.
_Claude Code on behalf of Claus Ibsen_
> 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
> Priority: Minor
> Fix For: 4.23.0
>
>
> 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)
