This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24139 in repository https://gitbox.apache.org/repos/asf/camel.git
commit 55fee1722c5d55707399992dcfd571a903d32a97 Author: Claus Ibsen <[email protected]> AuthorDate: Fri Jul 17 08:43:58 2026 +0200 CAMEL-24139: Fix Splitter watermark advancing past unrouted items on streaming abort Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../java/org/apache/camel/processor/Splitter.java | 22 ++++++++++------ .../camel/processor/SplitterWatermarkTest.java | 30 ++++++++++++++++++++++ 2 files changed, 44 insertions(+), 8 deletions(-) diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java index 2669a1652a5a..b047eb06c1e0 100644 --- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java +++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java @@ -290,6 +290,10 @@ public class Splitter extends MulticastProcessor { private final Exchange original; // tracks individual (raw) item count, independent of grouping private final AtomicInteger rawItemCount = new AtomicInteger(); + // tracks whether the primary (processing) iterator has been created; + // subsequent iterators (e.g. the drain in MulticastProcessor.doDone) + // must not update the watermark count (CAMEL-24139) + private boolean primaryIteratorCreated; private SplitterIterable(Exchange exchange, Object value) { this.original = exchange; @@ -349,6 +353,11 @@ public class Splitter extends MulticastProcessor { @Override public Iterator<ProcessorExchangePair> iterator() { + // only the first (primary) iterator tracks watermark count; + // subsequent iterators (drain in doDone) must not inflate it (CAMEL-24139) + boolean isPrimary = !primaryIteratorCreated; + primaryIteratorCreated = true; + return new Iterator<>() { private final Processor processor = getProcessors().iterator().next(); private int index; @@ -365,14 +374,6 @@ public class Splitter extends MulticastProcessor { if (!answer) { // we are now closed closed = true; - // store raw item count for watermark tracking (guard against - // re-iteration in MulticastProcessor.doDone which re-creates - // the iterator to release exchanges). - // Use rawItemCount (individual items) not index (which counts chunks when group > 0) - if (resumeStrategy != null && watermarkKey != null && watermarkExpression == null - && original.getProperty(SPLIT_WATERMARK_COUNT) == null) { - original.setProperty(SPLIT_WATERMARK_COUNT, rawItemCount.get()); - } // nothing more so we need to close the expression value in case it needs to be try { close(); @@ -422,6 +423,11 @@ public class Splitter extends MulticastProcessor { if (tracker != null) { tracker.incrementTotalItems(); } + // eagerly update watermark count for items actually routed (primary iterator only) + // so the drain loop in MulticastProcessor.doDone cannot inflate it (CAMEL-24139) + if (isPrimary && resumeStrategy != null && watermarkKey != null && watermarkExpression == null) { + original.setProperty(SPLIT_WATERMARK_COUNT, rawItemCount.get()); + } return createProcessorExchangePair(index++, processor, newExchange, route); } else { return null; diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterWatermarkTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterWatermarkTest.java index 490571ff3a22..d7f7947229bc 100644 --- a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterWatermarkTest.java +++ b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterWatermarkTest.java @@ -286,6 +286,24 @@ class SplitterWatermarkTest extends ContextTestSupport { "Watermark should not be updated when processing is aborted"); } + @Test + void testIndexBasedWatermarkNoUpdateOnHandledAbort() throws Exception { + // CAMEL-24139: streaming split + stopOnException + handled(true) should NOT + // advance the watermark past items that were never routed + MockEndpoint mock = getMockEndpoint("mock:handled-abort"); + mock.expectedBodiesReceived("a"); + + template.sendBody("direct:handled-abort", Arrays.asList("a", "FAIL", "c", "d", "e")); + + mock.assertIsSatisfied(); + + // items "a" (index 0) and "FAIL" (index 1) were both consumed and routed; + // watermark must cover them but NOT the unrouted items c, d, e + String watermark = store.get("handledAbortJob"); + assertNotNull(watermark, "Watermark should be set for routed items"); + assertEquals("1", watermark, "Watermark should cover items 0-1 (routed), not 0-4 (entire stream)"); + } + @Override protected RouteBuilder createRouteBuilder() { return new RouteBuilder() { @@ -373,6 +391,18 @@ class SplitterWatermarkTest extends ContextTestSupport { } }) .to("mock:abort-val-wm"); + + from("direct:handled-abort") + .onException(IllegalArgumentException.class).handled(true).end() + .split(body()).streaming().stopOnException() + .resumeStrategy(strategy, "handledAbortJob") + .process(exchange -> { + String body = exchange.getIn().getBody(String.class); + if ("FAIL".equals(body)) { + throw new IllegalArgumentException("Simulated failure"); + } + }) + .to("mock:handled-abort"); } }; }
