This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new f0a2b69bce24 CAMEL-24139: Fix Splitter watermark advancing past
unrouted items on streaming abort
f0a2b69bce24 is described below
commit f0a2b69bce249b637de98f456176e97d88a4bd94
Author: Claus Ibsen <[email protected]>
AuthorDate: Fri Jul 17 11:30:59 2026 +0200
CAMEL-24139: Fix Splitter watermark advancing past unrouted items on
streaming abort
Co-Authored-By: Claude Opus 4.6 <[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 7e154affc3eb..e6e7705c84c3 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();
@@ -425,6 +426,11 @@ public class Splitter extends MulticastProcessor {
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");
}
};
}
