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 7dad7adffd26 CAMEL-26710: camel-core - Split EIP: do not hang when the
watermark cannot be evaluated or stored (#29406)
7dad7adffd26 is described below
commit 7dad7adffd26874bce01d3b2b7fa35b9e2ed81d7
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Nov 8 16:31:21 2026 +0530
CAMEL-26710: camel-core - Split EIP: do not hang when the watermark cannot
be evaluated or stored (#29406)
* CAMEL-26710: camel-core - Split EIP: do not hang when the watermark
cannot be evaluated or stored
* CAMEL-26710: camel-core - Split EIP: upgrade guide entry for a watermark
that cannot be stored
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../org/apache/camel/catalog/docs/split-eip.adoc | 3 +-
.../main/docs/modules/eips/pages/split-eip.adoc | 3 +-
.../java/org/apache/camel/processor/Splitter.java | 31 +++--
.../SplitterWatermarkStoreFailureTest.java | 145 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 10 ++
5 files changed, 182 insertions(+), 10 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/split-eip.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/split-eip.adoc
index 43bc9ec864be..755fe4090c0d 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/split-eip.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/split-eip.adoc
@@ -1433,7 +1433,8 @@ NOTE: The `watermarkExpression` option uses the Simple
language for expression e
NOTE: The watermark is only updated when the split completes successfully. If
the split is aborted
(e.g., due to exceeding `maxFailedRecords`), the watermark is not updated,
which allows the
-failed batch to be retried.
+failed batch to be retried. The same applies when the `watermarkExpression`
fails, or when the resume
+strategy fails to store the watermark: the split fails with that exception,
and the items are processed again.
IMPORTANT: Watermark tracking assumes sequential route invocations (e.g.,
batch jobs triggered by
a `timer` or `scheduler`). If multiple exchanges hit the same route
concurrently, they will read
diff --git
a/core/camel-core-engine/src/main/docs/modules/eips/pages/split-eip.adoc
b/core/camel-core-engine/src/main/docs/modules/eips/pages/split-eip.adoc
index 43bc9ec864be..755fe4090c0d 100644
--- a/core/camel-core-engine/src/main/docs/modules/eips/pages/split-eip.adoc
+++ b/core/camel-core-engine/src/main/docs/modules/eips/pages/split-eip.adoc
@@ -1433,7 +1433,8 @@ NOTE: The `watermarkExpression` option uses the Simple
language for expression e
NOTE: The watermark is only updated when the split completes successfully. If
the split is aborted
(e.g., due to exceeding `maxFailedRecords`), the watermark is not updated,
which allows the
-failed batch to be retried.
+failed batch to be retried. The same applies when the `watermarkExpression`
fails, or when the resume
+strategy fails to store the watermark: the split fails with that exception,
and the items are processed again.
IMPORTANT: Watermark tracking assumes sequential route invocations (e.g.,
batch jobs triggered by
a `timer` or `scheduler`). If multiple exchanges hit the same route
concurrently, they will read
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 8cc7a7e896e2..0e2fb025bd47 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
@@ -234,13 +234,20 @@ public class Splitter extends MulticastProcessor {
return callback;
}
return doneSync -> {
- if (hasErrorThreshold) {
- buildSplitResult(exchange);
- }
- if (hasWatermark) {
- updateWatermark(exchange);
+ try {
+ if (hasErrorThreshold) {
+ buildSplitResult(exchange);
+ }
+ if (hasWatermark) {
+ updateWatermark(exchange);
+ }
+ } catch (Throwable e) {
+ // for example the resume strategy failed to store the
watermark
+ exchange.setException(e);
+ } finally {
+ // the callback must always be called, otherwise the exchange
never completes
+ callback.done(doneSync);
}
- callback.done(doneSync);
};
}
@@ -727,7 +734,14 @@ public class Splitter extends MulticastProcessor {
// evaluate watermark expression on completed sub-exchange (only
if no exception)
Exchange subExchange = delegate.getExchange();
if (subExchange.getException() == null) {
- String value = watermarkExpression.evaluate(subExchange,
String.class);
+ String value;
+ try {
+ value = watermarkExpression.evaluate(subExchange,
String.class);
+ } catch (Exception e) {
+ // fail the item, as an exception thrown here would make
the split exchange hang (never complete)
+ subExchange.setException(e);
+ return;
+ }
if (value != null) {
int itemIndex = delegate.getIndex();
// Use accumulateAndGet to keep the value from the
highest-indexed item.
@@ -780,12 +794,13 @@ public class Splitter extends MulticastProcessor {
}
private void persistWatermark(String value) {
- watermarkCache.put(watermarkKey, value);
try {
resumeStrategy.updateLastOffset(OffsetKeys.of(watermarkKey),
Offsets.of(value));
} catch (Exception e) {
throw RuntimeCamelException.wrapRuntimeCamelException(e);
}
+ // only cache the watermark once it is stored, so a failed exchange is
processed again
+ watermarkCache.put(watermarkKey, value);
}
private void buildSplitResult(Exchange exchange) {
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterWatermarkStoreFailureTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterWatermarkStoreFailureTest.java
new file mode 100644
index 000000000000..4bfa265e4a96
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterWatermarkStoreFailureTest.java
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.processor;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.RuntimeExpressionException;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.resume.Offset;
+import org.apache.camel.resume.OffsetKey;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * The split must complete, with the exception set, when the watermark cannot
be evaluated or stored.
+ */
+class SplitterWatermarkStoreFailureTest extends ContextTestSupport {
+
+ private final Map<String, String> store = new ConcurrentHashMap<>();
+ private final FailingResumeStrategy strategy = new
FailingResumeStrategy(store);
+
+ @Test
+ void testSplitCompletesWhenStoringTheWatermarkFails() throws Exception {
+ strategy.failures.set(1);
+ getMockEndpoint("mock:split").expectedMessageCount(3);
+
+ Exchange out = template.asyncSend("direct:start", e ->
e.getIn().setBody(Arrays.asList("a", "b", "c")))
+ .get(5, TimeUnit.SECONDS);
+
+ MockEndpoint.assertIsSatisfied(context);
+ assertNotNull(out.getException());
+ assertInstanceOf(IllegalStateException.class,
out.getException().getCause());
+ assertFalse(store.containsKey("failJob"));
+ }
+
+ @Test
+ void testFailedWatermarkIsNotUsedByTheNextSplit() throws Exception {
+ strategy.failures.set(1);
+ Exchange out = template.asyncSend("direct:start", e ->
e.getIn().setBody(Arrays.asList("a", "b", "c")))
+ .get(5, TimeUnit.SECONDS);
+ assertNotNull(out.getException());
+
+ // the store works again: the items whose watermark was not stored are
processed again
+ MockEndpoint mock = getMockEndpoint("mock:split");
+ mock.reset();
+ mock.expectedBodiesReceived("a", "b", "c");
+
+ out = template.asyncSend("direct:start", e ->
e.getIn().setBody(Arrays.asList("a", "b", "c")))
+ .get(5, TimeUnit.SECONDS);
+
+ mock.assertIsSatisfied();
+ assertNull(out.getException());
+ assertEquals("2", store.get("failJob"));
+ }
+
+ @Test
+ void testRedeliveryProcessesTheItemsAgainWhenStoringTheWatermarkFails()
throws Exception {
+ strategy.failures.set(1);
+ getMockEndpoint("mock:redeliver").expectedBodiesReceived("a", "b",
"c", "a", "b", "c");
+
+ Exchange out = template.asyncSend("direct:redeliver", e ->
e.getIn().setBody(Arrays.asList("a", "b", "c")))
+ .get(5, TimeUnit.SECONDS);
+
+ MockEndpoint.assertIsSatisfied(context);
+ assertNull(out.getException());
+ assertEquals("2", store.get("redeliverJob"));
+ }
+
+ @Test
+ void testSplitCompletesWhenTheWatermarkExpressionFails() throws Exception {
+ getMockEndpoint("mock:value").expectedMessageCount(2);
+
+ Exchange out = template.asyncSend("direct:value", e ->
e.getIn().setBody(Arrays.asList("a", "b")))
+ .get(5, TimeUnit.SECONDS);
+
+ MockEndpoint.assertIsSatisfied(context);
+ assertInstanceOf(RuntimeExpressionException.class, out.getException());
+ assertFalse(store.containsKey("valueJob"));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .split(body()).resumeStrategy(strategy, "failJob")
+ .to("mock:split");
+
+ from("direct:redeliver")
+
.errorHandler(defaultErrorHandler().maximumRedeliveries(1).redeliveryDelay(0))
+ .split(body()).resumeStrategy(strategy, "redeliverJob")
+ .to("mock:redeliver");
+
+ // the watermark expression fails on every item
+ from("direct:value")
+ .split(body()).resumeStrategy(strategy,
"valueJob").watermarkExpression("${body.noSuchMethod()}")
+ .to("mock:value");
+ }
+ };
+ }
+
+ private static class FailingResumeStrategy extends
SplitterTestResumeStrategy {
+
+ private final AtomicInteger failures = new AtomicInteger();
+
+ FailingResumeStrategy(Map<String, String> store) {
+ super(store);
+ }
+
+ @Override
+ public void updateLastOffset(OffsetKey<?> offsetKey, Offset<?>
offsetValue) {
+ if (failures.getAndDecrement() > 0) {
+ throw new IllegalStateException("Cannot store the watermark");
+ }
+ super.updateLastOffset(offsetKey, offsetValue);
+ }
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 790e3865ee8b..084e3f22476b 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -669,6 +669,16 @@ now receive the input that was split, instead of that one
split item. A route th
read the `CamelSplitResult` exchange property, set as before:
`SplitResult.getFailures()` has the index and the
exception of the failed items.
+=== camel-core - Split EIP processes the items again when the watermark cannot
be stored
+
+When the resume strategy of a split with watermark tracking fails to store the
watermark, the split now fails with that
+exception, and the next split processes the same items again. Before, with
`synchronous()`, the split also failed, but
+the watermark was kept in memory anyway, so the next split skipped those items
although the watermark was never stored.
+Without `synchronous()`, the exchange never completed (fixed by the same
change).
+
+A `watermarkExpression` that throws on an item now fails that item, so
`stopOnException`, `maxFailedRecords` and
+`errorThreshold` apply as for any failed item.
+
=== camel-core - shareUnitOfWork: a part sent to seda no longer takes the on
completions of the parent exchange
Since Camel 4.14 the parts of a Split, Multicast or Recipient List with
`shareUnitOfWork` use the unit of work of the
