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 8bd61afc3b7d CAMEL-26321: camel-core - Split EIP keeps the original
message when maxFailedRecords or errorThreshold stops it (#28781)
8bd61afc3b7d is described below
commit 8bd61afc3b7d7b8ffc7e34502d73fb766412acd6
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Nov 1 13:49:56 2026 +0530
CAMEL-26321: camel-core - Split EIP keeps the original message when
maxFailedRecords or errorThreshold stops it (#28781)
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 +
.../apache/camel/processor/MulticastProcessor.java | 18 +-
.../java/org/apache/camel/processor/Splitter.java | 21 ++
...itterThresholdStopKeepsOriginalMessageTest.java | 280 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 12 +
6 files changed, 335 insertions(+), 2 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 df5289871bbb..db0d1f2ae172 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
@@ -1263,6 +1263,9 @@ In this example, if 50% or more of the processed items
have failed, the splitter
Both `maxFailedRecords` and `errorThreshold` can be combined. The splitter
stops when either
threshold is exceeded.
+When the splitter stops because a threshold is reached, the exchange keeps the
original input message
+and has the exception set, the same as with `stopOnException`.
+
IMPORTANT: The `stopOnException` option is mutually exclusive with
`maxFailedRecords` and
`errorThreshold`. You cannot use `stopOnException` together with either of
these options.
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 df5289871bbb..db0d1f2ae172 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
@@ -1263,6 +1263,9 @@ In this example, if 50% or more of the processed items
have failed, the splitter
Both `maxFailedRecords` and `errorThreshold` can be combined. The splitter
stops when either
threshold is exceeded.
+When the splitter stops because a threshold is reached, the exchange keeps the
original input message
+and has the exception set, the same as with `stopOnException`.
+
IMPORTANT: The `stopOnException` option is mutually exclusive with
`maxFailedRecords` and
`errorThreshold`. You cannot use `stopOnException` together with either of
these options.
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java
index 9b54724c468b..f2ede9dc193a 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java
@@ -875,6 +875,19 @@ public class MulticastProcessor extends
BaseProcessorSupport
return !stopOnException;
}
+ /**
+ * Whether processing was stopped because {@link
#shouldContinueOnFailure(Exchange, Exchange, int)} returned
+ * {@code false} for a failed sub-exchange although {@code
stopOnException} is not enabled (e.g., the Splitter
+ * reached its error threshold). The original exchange is then kept and
only gets the exception, the same as with
+ * {@code stopOnException}.
+ *
+ * @param original the original exchange
+ * @return {@code true} if processing was stopped on a failure
+ */
+ protected boolean isStoppedOnFailure(Exchange original) {
+ return false;
+ }
+
protected ScheduledFuture<?> schedule(Executor executor, Runnable
runnable, long delay, TimeUnit unit) {
if (executor instanceof ScheduledExecutorService
scheduledExecutorService) {
return scheduledExecutorService.schedule(runnable, delay, unit);
@@ -946,7 +959,8 @@ public class MulticastProcessor extends BaseProcessorSupport
// cleanup any per exchange aggregation strategy
removeAggregationStrategyFromExchange(original);
- // we need to know if there was an exception, and if the
stopOnException option was enabled
+ // we need to know if there was an exception, and if we stopped on it
(the stopOnException option, or a failure
+ // policy such as the error threshold of the Splitter)
// also we would need to know if any error handler has attempted
redelivery and exhausted
boolean stoppedOnException = false;
boolean exception = false;
@@ -954,7 +968,7 @@ public class MulticastProcessor extends BaseProcessorSupport
&& (subExchange.getException() != null ||
subExchange.getExchangeExtension().isRedeliveryExhausted());
if (original.getException() != null || subExchange != null &&
subExchange.getException() != null) {
// there was an exception and we stopped
- stoppedOnException = isStopOnException();
+ stoppedOnException = isStopOnException() ||
isStoppedOnFailure(original);
exception = true;
}
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 c00f2c13045f..c3ab307b41e1 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
@@ -596,6 +596,7 @@ public class Splitter extends MulticastProcessor {
// check if we've exceeded the max failed records
if (maxFailedRecords > 0 && tracker.getFailureCount() >=
maxFailedRecords) {
+ tracker.markStopped();
return false;
}
// check if we've exceeded the error ratio threshold
@@ -604,6 +605,7 @@ public class Splitter extends MulticastProcessor {
if (processed > 0) {
double ratio = (double) tracker.getFailureCount() /
processed;
if (ratio >= errorThreshold) {
+ tracker.markStopped();
return false;
}
}
@@ -615,11 +617,22 @@ public class Splitter extends MulticastProcessor {
return true;
}
+ @Override
+ protected boolean isStoppedOnFailure(Exchange original) {
+ // the error threshold of THIS splitter was reached, so keep the
original exchange as stopOnException does
+ if (errorThreshold <= 0 && maxFailedRecords <= 0) {
+ return false;
+ }
+ SplitFailureTracker tracker =
original.getProperty(SPLIT_FAILURE_TRACKER, SplitFailureTracker.class);
+ return tracker != null && tracker.isStopped();
+ }
+
static final class SplitFailureTracker {
private final AtomicInteger failureCount = new AtomicInteger();
private final AtomicInteger totalItems = new AtomicInteger();
private final AtomicInteger processedItems = new AtomicInteger();
private final CopyOnWriteArrayList<SplitResult.Failure> failures = new
CopyOnWriteArrayList<>();
+ private volatile boolean stopped;
void recordFailure(int index, Exception exception) {
failureCount.incrementAndGet();
@@ -646,6 +659,14 @@ public class Splitter extends MulticastProcessor {
return failureCount.get();
}
+ void markStopped() {
+ stopped = true;
+ }
+
+ boolean isStopped() {
+ return stopped;
+ }
+
List<SplitResult.Failure> getFailures() {
return List.copyOf(failures);
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterThresholdStopKeepsOriginalMessageTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterThresholdStopKeepsOriginalMessageTest.java
new file mode 100644
index 000000000000..e198d548974b
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterThresholdStopKeepsOriginalMessageTest.java
@@ -0,0 +1,280 @@
+/*
+ * 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.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+
+import org.apache.camel.AggregationStrategy;
+import org.apache.camel.CamelExchangeException;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.SplitResult;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * When the Splitter stops because maxFailedRecords or errorThreshold is
reached, the exchange keeps the original input
+ * message and gets the exception, the same as with stopOnException, instead
of the message of the failed split item.
+ */
+class SplitterThresholdStopKeepsOriginalMessageTest extends ContextTestSupport
{
+
+ private final RecordingAggregationStrategy stopStrategy = new
RecordingAggregationStrategy();
+ private final RecordingAggregationStrategy maxStrategy = new
RecordingAggregationStrategy();
+
+ private Exchange send(String uri, Object body) {
+ return send(uri, body, false);
+ }
+
+ private Exchange send(String uri, Object body, boolean transacted) {
+ return template.send(uri, e -> {
+ e.getMessage().setBody(body);
+ e.getMessage().setHeader("orig", "yes");
+ e.setProperty("origProp", "yes");
+ e.getExchangeExtension().setTransacted(transacted);
+ });
+ }
+
+ private static void assertOriginalKept(Exchange out, String body) {
+ assertInstanceOf(IllegalStateException.class,
out.getException().getCause());
+ assertOriginalKept(out, (Object) body);
+ }
+
+ private static void assertOriginalKept(Exchange out, Object body) {
+ assertInstanceOf(CamelExchangeException.class, out.getException());
+ assertEquals(body, out.getMessage().getBody(), "body of the original
message");
+ assertEquals("yes", out.getMessage().getHeader("orig"));
+ assertNull(out.getMessage().getHeader("item"), "header of the failed
split item");
+ assertNull(out.getProperty("itemProp"), "property of the failed split
item");
+ assertNull(out.getVariable("itemVar"), "variable of the failed split
item");
+ assertEquals("yes", out.getProperty("origProp"), "property removed in
the failed split item");
+ }
+
+ @Test
+ void stopOnExceptionKeepsOriginal() {
+ assertOriginalKept(send("direct:stop", "a,b,c"), "a,b,c");
+ }
+
+ @Test
+ void maxFailedRecordsKeepsOriginal() {
+ Exchange out = send("direct:max", "a,b,c");
+ assertOriginalKept(out, "a,b,c");
+ assertTrue(out.getProperty(Exchange.SPLIT_RESULT,
SplitResult.class).isAborted());
+ }
+
+ @Test
+ void maxFailedRecordsAfterToleratedFailureKeepsOriginal() {
+ Exchange out = send("direct:max2", "a,b,c");
+ assertOriginalKept(out, "a,b,c");
+ // the split result still has the index and the exception of each
failed item
+ SplitResult result = out.getProperty(Exchange.SPLIT_RESULT,
SplitResult.class);
+ assertTrue(result.isAborted());
+ assertEquals(List.of(0, 1),
result.getFailures().stream().map(SplitResult.Failure::index).toList());
+ assertInstanceOf(IllegalStateException.class,
result.getFailures().get(1).exception());
+ }
+
+ @Test
+ void shareUnitOfWorkMaxFailedRecordsKeepsOriginal() {
+ assertOriginalKept(send("direct:share", "a,b,c"), "a,b,c");
+ }
+
+ @Test
+ void nestedSplitKeepsOriginalOfOuterSplit() {
+ // only the outer split has a threshold, the inner split fails the
outer item
+ List<List<String>> body = List.of(List.of("a", "FAIL"), List.of("b"));
+ Exchange out = send("direct:nested", body);
+ assertOriginalKept(out, body);
+ assertNull(out.getMessage().getHeader("outerItem"), "header of the
failed outer split item");
+ }
+
+ @Test
+ void nestedSplitsWithThresholdsKeepOriginalOfOuterSplit() {
+ List<List<String>> body = List.of(List.of("a", "FAIL"), List.of("b"));
+ Exchange out = send("direct:nestedBoth", body);
+ assertOriginalKept(out, body);
+ assertNull(out.getMessage().getHeader("outerItem"), "header of the
failed outer split item");
+ }
+
+ @Test
+ void errorThresholdKeepsOriginal() {
+ Exchange out = send("direct:ratio", "a,b,c");
+ assertOriginalKept(out, "a,b,c");
+ assertTrue(out.getProperty(Exchange.SPLIT_RESULT,
SplitResult.class).isAborted());
+ }
+
+ @Test
+ void parallelMaxFailedRecordsKeepsOriginal() {
+ assertOriginalKept(send("direct:parallel", "a,b,c"), "a,b,c");
+ }
+
+ @Test
+ void streamingMaxFailedRecordsKeepsOriginal() {
+ assertOriginalKept(send("direct:streaming", "a,b,c"), "a,b,c");
+ }
+
+ @Test
+ void transactedMaxFailedRecordsKeepsOriginal() {
+ // the transacted code path (MulticastTransactedTask)
+ assertOriginalKept(send("direct:max", "a,b,c", true), "a,b,c");
+ }
+
+ @Test
+ void sameExceptionAndPropertiesAsStopOnException() {
+ Exchange stop = send("direct:stop", "a,b,c");
+ Exchange max = send("direct:max", "a,b,c");
+
+ // the same exception, which says which split item failed (and the id
of its exchange)
+ assertTrue(stop.getException().getMessage().startsWith("Multicast
processing failed for number 0."));
+ assertTrue(max.getException().getMessage().startsWith("Multicast
processing failed for number 0."));
+ assertEquals(stop.getException().getClass(),
max.getException().getClass());
+ assertEquals(stop.getException().getCause().getClass(),
max.getException().getCause().getClass());
+ assertEquals(stop.getException().getCause().getMessage(),
max.getException().getCause().getMessage());
+ assertEquals(stop.getExchangeExtension().isRedeliveryExhausted(),
max.getExchangeExtension().isRedeliveryExhausted());
+ Set<String> stopProps = new TreeSet<>(stop.getProperties().keySet());
+ Set<String> maxProps = new TreeSet<>(max.getProperties().keySet());
+ // only the threshold options set the split result
+ maxProps.remove(Exchange.SPLIT_RESULT);
+ assertEquals(stopProps, maxProps);
+ }
+
+ @Test
+ void doCatchAroundSplitGetsOriginal() throws Exception {
+ MockEndpoint error = getMockEndpoint("mock:error");
+ error.expectedBodiesReceived("a,b,c");
+ error.expectedHeaderReceived("orig", "yes");
+ error.message(0).header("item").isNull();
+
+ send("direct:trycatch", "a,b,c");
+
+ error.assertIsSatisfied();
+ }
+
+ @Test
+ void doCatchAroundStopOnExceptionSplitGetsOriginal() throws Exception {
+ MockEndpoint error = getMockEndpoint("mock:error");
+ error.expectedBodiesReceived("a,b,c");
+ error.expectedHeaderReceived("orig", "yes");
+ error.message(0).header("item").isNull();
+
+ send("direct:trycatchstop", "a,b,c");
+
+ error.assertIsSatisfied();
+ }
+
+ @Test
+ void aggregationStrategySeesTheStopAsWithStopOnException() {
+ Exchange stop = send("direct:aggstop", "ok,FAIL,ok");
+ Exchange max = send("direct:aggmax", "ok,FAIL,ok");
+
+ assertOriginalKept(stop, "ok,FAIL,ok");
+ assertOriginalKept(max, "ok,FAIL,ok");
+ // the item before the failed one is aggregated, the failed one is
only given to onCompletion
+ assertEquals(List.of("ok"), stopStrategy.aggregated);
+ assertEquals(stopStrategy.aggregated, maxStrategy.aggregated);
+ assertNotNull(stopStrategy.completedWith);
+ assertNotNull(maxStrategy.completedWith);
+ assertInstanceOf(CamelExchangeException.class,
maxStrategy.completedWith.getException());
+ assertEquals("FAIL",
maxStrategy.completedWith.getMessage().getHeader("item"));
+ assertEquals(stopStrategy.completedWith.getMessage().getHeader("item"),
+ maxStrategy.completedWith.getMessage().getHeader("item"));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:stop").split(body()).stopOnException().to("direct:item").end();
+
from("direct:max").split(body()).maxFailedRecords(1).to("direct:item").end();
+
from("direct:max2").split(body()).maxFailedRecords(2).to("direct:item").end();
+
from("direct:ratio").split(body()).errorThreshold(0.5).to("direct:item").end();
+
from("direct:parallel").split(body()).parallelProcessing().maxFailedRecords(1).to("direct:item").end();
+
from("direct:streaming").split(body()).streaming().maxFailedRecords(1).to("direct:item").end();
+
from("direct:share").split(body()).shareUnitOfWork().maxFailedRecords(1).to("direct:item").end();
+
+ from("direct:nested").split(body()).maxFailedRecords(1)
+ .setHeader("outerItem", simple("${body}"))
+ .split(body()).to("direct:failing").end()
+ .end();
+ from("direct:nestedBoth").split(body()).maxFailedRecords(1)
+ .setHeader("outerItem", simple("${body}"))
+
.split(body()).maxFailedRecords(1).to("direct:failing").end()
+ .end();
+
+ from("direct:trycatch")
+ .doTry()
+
.split(body()).maxFailedRecords(1).to("direct:item").end()
+ .endDoTry()
+ .doCatch(Exception.class)
+ .to("mock:error")
+ .end();
+ from("direct:trycatchstop")
+ .doTry()
+
.split(body()).stopOnException().to("direct:item").end()
+ .endDoTry()
+ .doCatch(Exception.class)
+ .to("mock:error")
+ .end();
+
+ from("direct:aggstop").split(body(),
stopStrategy).stopOnException().to("direct:failing").end();
+ from("direct:aggmax").split(body(),
maxStrategy).maxFailedRecords(1).to("direct:failing").end();
+
+ from("direct:item")
+ .setHeader("item", simple("${body}"))
+ .setProperty("itemProp", constant("x"))
+ .removeProperty("origProp")
+ .setVariable("itemVar", constant("x"))
+ .setBody(constant("changed"))
+ .throwException(new IllegalStateException("boom"));
+
+ from("direct:failing")
+ .setHeader("item", simple("${body}"))
+ .setProperty("itemProp", constant("x"))
+ .removeProperty("origProp")
+ .setVariable("itemVar", constant("x"))
+ .filter(body().isEqualTo("FAIL"))
+ .throwException(new IllegalStateException("boom"));
+ }
+ };
+ }
+
+ private static final class RecordingAggregationStrategy implements
AggregationStrategy {
+
+ private final List<String> aggregated = new ArrayList<>();
+ private volatile Exchange completedWith;
+
+ @Override
+ public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
+ aggregated.add(newExchange.getMessage().getBody(String.class));
+ return newExchange;
+ }
+
+ @Override
+ public void onCompletion(Exchange exchange, Exchange inputExchange) {
+ completedWith = exchange;
+ }
+ }
+}
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 3dd6687f2d48..eda9a6370f42 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
@@ -272,6 +272,18 @@ An exception thrown by the `isReadyToBeClosed` method of a
`halfOpenHandler` is
the policy), and the circuit is kept open and checked again after
`halfOpenAfter`; before, the timer thread died and the
circuit stayed open.
+=== camel-core - Split EIP keeps the original message when maxFailedRecords or
errorThreshold stops it
+
+When the Split EIP stops because `maxFailedRecords` or `errorThreshold` is
reached, the exchange now keeps its original
+input message, headers, exchange properties and variables, and only gets the
exception, as with `stopOnException`.
+Before, the message, headers, exchange properties and variables of the split
item that reached the threshold, as the
+steps inside the split had changed them, were copied onto the exchange.
+
+So a `doCatch` around the split, an `onException`, an `onCompletion`, a dead
letter channel or the caller of the route
+now receive the input that was split, instead of that one split item. A route
that needs to know which items failed can
+read the `CamelSplitResult` exchange property, set as before:
`SplitResult.getFailures()` has the index and the
+exception of the failed items.
+
=== camel-main - a failure to start is logged
A Camel Main application that fails to start now logs why, at ERROR: `Error
starting Camel: ...` with the stack trace,
