[jira] [Updated] (CAMEL-24139) camel-core - Splitter watermark advances past unprocessed items when a streaming split aborts without an exception (data loss on resume)

["Federico Mariani (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/CAMEL-24139?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Federico Mariani updated CAMEL-24139:
-------------------------------------
    Attachment: SplitterWatermarkAdvancesOnAbortIssueTest.java

> camel-core - Splitter watermark advances past unprocessed items when a streaming split aborts without an exception (data loss on resume)
> ----------------------------------------------------------------------------------------------------------------------------------------
>
>                 Key: CAMEL-24139
>                 URL: https://issues.apache.org/jira/browse/CAMEL-24139
>             Project: Camel
>          Issue Type: Bug
>          Components: camel-core, eip
>    Affects Versions: 4.22.0
>            Reporter: Federico Mariani
>            Priority: Critical
>         Attachments: SplitterWatermarkAdvancesOnAbortIssueTest.java
>
>
> With the new index-based watermark support (CAMEL-23264, unreleased 4.22.0-SNAPSHOT), when a *streaming* split stops early, {{MulticastProcessor.doDone()}} re-iterates the streaming {{SplitterIterable}} to release the pooled exchanges ({{MulticastProcessor.java:911-920}}). That drain drives the counting iterator to exhaustion, so {{hasNext()}} stores the *full raw item count* in {{CamelSplitWatermarkCount}} ({{Splitter.java:368-375}}) — including items that were never routed.
> {{Splitter.updateWatermark()}} only refuses to persist when {{exchange.getException() != null}}, so any abort path that leaves the original exchange without an exception persists a watermark covering the whole stream:
> * {{stopOnException}} + the exception *handled* by the error handler ({{handled(true)}}) — the stop path sets {{result.set(exchange)}} with no exception ({{MulticastProcessor.java:625}});
> * {{parallelProcessing}} + {{timeout}} — the timeout path completes the multicast without any exception, while unfinished/unsubmitted items are drained.
> On the next run every unprocessed item is skipped: *silent data loss* in exactly the incremental-processing scenario the watermark feature targets.
> *Reproducer (attached, fails on main):* {{SplitterWatermarkAdvancesOnAbortIssueTest}} — 5 items, {{streaming()}} + {{stopOnException()}} + {{onException(...).handled(true)}}, abort at index 1; the stored watermark is {{"4"}} although only item 0 was processed:
> {noformat}
> org.opentest4j.AssertionFailedError: Watermark advanced past items that were never processed (processed=[a]) ==> expected: not equal but was: <4>
> {noformat}
> (The test reuses the {{SplitterTestResumeStrategy}} helper already in the tree from CAMEL-23264.)
> Suggested fix direction: capture the raw item count when processing stops (e.g. only count items handed out while the multicast task is not done), or mark the iterable as aborted before the release drain in {{doDone}} and skip the watermark-count capture during the drain.
> Since CAMEL-23264 is unreleased, this can still be fixed before 4.22.0 ships.
> _Filed by Claude Code on behalf of Federico Mariani (fmariani)_



--
This message was sent by Atlassian Jira
(v8.20.10#820010)