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 b1ebadeffa08 CAMEL-26476: camel-core - Split EIP: a nested split
corrupts the processed items of the outer split (#28820)
b1ebadeffa08 is described below
commit b1ebadeffa08d2d93b5951771478cf4b4ca4c4a1
Author: Adriano Machado <[email protected]>
AuthorDate: Mon Nov 2 01:30:10 2026 -0500
CAMEL-26476: camel-core - Split EIP: a nested split corrupts the processed
items of the outer split (#28820)
…essed items of the outer split
Splitter.afterSend read the failure tracker from the properties of the item
exchange, which inherits them, without checking that this splitter has
thresholds. An inner split without thresholds counted its items in the
tracker
of the outer split, and one with thresholds removed that tracker from the
exchange of the outer item. errorThreshold divides by the processed items,
so
the outer split did not stop when it should.
afterSend now gets the original exchange, as shouldContinueOnFailure does,
and
has the same check as the other places that use the tracker.
_Claude Code on behalf of Adriano Machado (@ammachado)_
_This was generated by an AI agent and may contain inaccuracies.
Please verify before relying on it._
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/processor/MulticastProcessor.java | 16 ++++-
.../java/org/apache/camel/processor/Splitter.java | 13 ++--
.../SplitterNestedFailureTrackerLeakTest.java | 73 ++++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 7 +++
4 files changed, 102 insertions(+), 7 deletions(-)
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 f2ede9dc193a..7966de01cc42 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
@@ -680,7 +680,7 @@ public class MulticastProcessor extends BaseProcessorSupport
StopWatch watch = beforeSend(pair);
AsyncCallback taskCallback = (doneSync) -> {
- afterSend(pair, watch);
+ afterSend(pair, watch, original);
// Decide whether to continue with the multicast or
not; similar logic to the Pipeline
// remember to test for stop on exception and
aggregate before copying back results
@@ -813,7 +813,7 @@ public class MulticastProcessor extends BaseProcessorSupport
} catch (Exception e) {
exchange.setException(e);
} finally {
- afterSend(pair, watch);
+ afterSend(pair, watch, original);
}
// Decide whether to continue with the multicast or not; similar
logic to the Pipeline
@@ -921,6 +921,18 @@ public class MulticastProcessor extends
BaseProcessorSupport
return watch;
}
+ /**
+ * Called when an item was sent.
+ *
+ * @param pair the pair of the item
+ * @param watch the watch from {@link
#beforeSend(ProcessorExchangePair)}
+ * @param original the exchange that was split or multicasted. Properties
should be read from this exchange, as an
+ * item can have inherited them from an outer split, or a
nested split can have changed them.
+ */
+ protected void afterSend(ProcessorExchangePair pair, StopWatch watch,
Exchange original) {
+ afterSend(pair, watch);
+ }
+
protected void afterSend(ProcessorExchangePair pair, StopWatch watch) {
// we are done with the exchange pair
pair.done();
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 c3ab307b41e1..b13ffd8aed62 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
@@ -562,11 +562,14 @@ public class Splitter extends MulticastProcessor {
}
@Override
- protected void afterSend(ProcessorExchangePair pair, StopWatch watch) {
- super.afterSend(pair, watch);
- SplitFailureTracker tracker =
pair.getExchange().getProperty(SPLIT_FAILURE_TRACKER,
SplitFailureTracker.class);
- if (tracker != null) {
- tracker.incrementProcessedItems();
+ protected void afterSend(ProcessorExchangePair pair, StopWatch watch,
Exchange original) {
+ super.afterSend(pair, watch, original);
+ // only when THIS splitter has thresholds configured to avoid
corrupting an outer tracker
+ if (errorThreshold > 0 || maxFailedRecords > 0) {
+ SplitFailureTracker tracker =
original.getProperty(SPLIT_FAILURE_TRACKER, SplitFailureTracker.class);
+ if (tracker != null) {
+ tracker.incrementProcessedItems();
+ }
}
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterNestedFailureTrackerLeakTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterNestedFailureTrackerLeakTest.java
index baf416580b3e..c087263f484e 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterNestedFailureTrackerLeakTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterNestedFailureTrackerLeakTest.java
@@ -20,10 +20,12 @@ import java.util.List;
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.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
@@ -77,6 +79,53 @@ class SplitterNestedFailureTrackerLeakTest extends
ContextTestSupport {
"Inner split absorbs its failure within threshold, outer
should succeed");
}
+ @Test
+ void testInnerSplitWithoutThresholdIsNotCountedByOuterSplit() {
+ // the 3 items of each inner split are not items of the outer split
+ Exchange result = template.send("direct:outerMaxFailed",
+ e -> e.getIn().setBody(List.of(List.of("a", "b", "c"),
List.of("d", "e", "f"))));
+
+ SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT,
SplitResult.class);
+ assertEquals(2, splitResult.getTotalItems());
+ assertEquals(2, splitResult.getProcessedItems());
+ }
+
+ @Test
+ void testInnerSplitWithThresholdDoesNotHideOuterItems() {
+ // an inner split with its own tracker removes it from the exchange of
the outer item when it completes
+ Exchange result = template.send("direct:bothThresholds",
+ e -> e.getIn().setBody(List.of(List.of("a", "b", "c"),
List.of("d", "e", "f"))));
+
+ SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT,
SplitResult.class);
+ assertEquals(2, splitResult.getTotalItems());
+ assertEquals(2, splitResult.getProcessedItems());
+ }
+
+ @Test
+ void testOuterErrorThresholdIsNotDilutedByInnerItems() throws Exception {
+ // 1 failure of 2 outer items reaches the threshold of 0.5, so the
third outer item must not run. Counting the
+ // 9 items of the inner split would make that 1 of 11, and the outer
split would continue
+ MockEndpoint mock = getMockEndpoint("mock:outerRatio");
+ mock.expectedMessageCount(9);
+
+ template.send("direct:outerRatio", e -> e.getIn().setBody(List.of(
+ List.of("1", "2", "3", "4", "5", "6", "7", "8", "9"),
List.of("FAIL_INNER"), List.of("not reached"))));
+
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void testOuterErrorThresholdWorksWithInnerThreshold() throws Exception {
+ // without processed items the ratio is never computed, so the outer
split would never stop
+ MockEndpoint mock = getMockEndpoint("mock:outerRatioInnerThreshold");
+ mock.expectedMessageCount(1);
+
+ template.send("direct:outerRatioInnerThreshold", e ->
e.getIn().setBody(List.of(
+ List.of("1"), List.of("FAIL_INNER"), List.of("not reached"))));
+
+ mock.assertIsSatisfied();
+ }
+
@Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
@@ -96,6 +145,30 @@ class SplitterNestedFailureTrackerLeakTest extends
ContextTestSupport {
.end()
.end();
+ from("direct:outerRatio")
+ .split(body()).errorThreshold(0.5)
+ .split(body())
+ .process(exchange -> {
+ if
("FAIL_INNER".equals(exchange.getIn().getBody(String.class))) {
+ throw new IllegalArgumentException("Inner
failure");
+ }
+ })
+ .to("mock:outerRatio")
+ .end()
+ .end();
+
+ from("direct:outerRatioInnerThreshold")
+ .split(body()).errorThreshold(0.5)
+ .split(body()).maxFailedRecords(1)
+ .process(exchange -> {
+ if
("FAIL_INNER".equals(exchange.getIn().getBody(String.class))) {
+ throw new IllegalArgumentException("Inner
failure");
+ }
+ })
+ .to("mock:outerRatioInnerThreshold")
+ .end()
+ .end();
+
// Both outer and inner with maxFailedRecords
from("direct:bothThresholds")
.split(body()).maxFailedRecords(1)
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 b50ad720d318..a90914d81fb9 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
@@ -107,6 +107,13 @@ at most once per millisecond, so that the JMX statistics
no longer generate an e
several exchanges complete within the same millisecond, the attribute has the
id of the first one, not of the last one.
Failed exchanges are not affected: `LastExchangeFailureExchangeId` is still
updated for every failed exchange.
+=== camel-core - a nested split no longer changes the processed items of the
outer split
+
+With `errorThreshold` or `maxFailedRecords` on a split that contains another
split, the inner split changed the number of
+processed items of the outer split. It counted its own items when it had no
thresholds, and hid the items of the outer
+split when it had them. `SplitResult.getProcessedItems()` now reports the
items of that split only. A nested split with
+`errorThreshold` can now stop earlier than before, as the ratio is computed
from the correct number.
+
=== camel-core - RedeliveryErrorHandler.defensiveCopyExchangeIfNeeded is
deprecated
When redelivery is enabled, the error handler (`DefaultErrorHandler`,
`DeadLetterChannel` and the transaction error
