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 10065b6f7242 CAMEL-24143: Fix medium findings in Split & Aggregate EIP
processors
10065b6f7242 is described below
commit 10065b6f72424a8f46ab85881021720a2d171358
Author: Claus Ibsen <[email protected]>
AuthorDate: Sat Jul 18 11:44:07 2026 +0200
CAMEL-24143: Fix medium findings in Split & Aggregate EIP processors
Co-Authored-By: Claude Opus 4.6 <[email protected]>
---
.../main/java/org/apache/camel/SplitResult.java | 16 ++++-
.../java/org/apache/camel/processor/Splitter.java | 73 +++++++++++++++++-----
.../processor/aggregate/AggregateProcessor.java | 12 +++-
.../processor/SplitterErrorThresholdTest.java | 4 +-
.../camel/processor/SplitterSplitResultTest.java | 3 +-
5 files changed, 84 insertions(+), 24 deletions(-)
diff --git a/core/camel-api/src/main/java/org/apache/camel/SplitResult.java
b/core/camel-api/src/main/java/org/apache/camel/SplitResult.java
index 6dbf371ef6ba..a8d125642be8 100644
--- a/core/camel-api/src/main/java/org/apache/camel/SplitResult.java
+++ b/core/camel-api/src/main/java/org/apache/camel/SplitResult.java
@@ -40,12 +40,14 @@ public final class SplitResult {
}
private final int totalItems;
+ private final int processedItems;
private final int failureCount;
private final List<Failure> failures;
private final boolean aborted;
- public SplitResult(int totalItems, int failureCount, List<Failure>
failures, boolean aborted) {
+ public SplitResult(int totalItems, int processedItems, int failureCount,
List<Failure> failures, boolean aborted) {
this.totalItems = totalItems;
+ this.processedItems = processedItems;
this.failureCount = failureCount;
this.failures = failures != null ?
Collections.unmodifiableList(failures) : Collections.emptyList();
this.aborted = aborted;
@@ -59,11 +61,20 @@ public final class SplitResult {
return totalItems;
}
+ /**
+ * The number of items that were actually processed. After an abort, this
may be less than {@code getTotalItems()}.
+ *
+ * @since 4.22
+ */
+ public int getProcessedItems() {
+ return processedItems;
+ }
+
/**
* The number of items that completed successfully.
*/
public int getSuccessCount() {
- return totalItems - failureCount;
+ return processedItems - failureCount;
}
/**
@@ -90,6 +101,7 @@ public final class SplitResult {
@Override
public String toString() {
return "SplitResult[total=" + totalItems
+ + ", processed=" + processedItems
+ ", success=" + getSuccessCount()
+ ", failures=" + failureCount
+ ", aborted=" + aborted + "]";
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 e6e7705c84c3..78a89bdc33ba 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
@@ -55,6 +55,7 @@ import org.apache.camel.support.resume.OffsetKeys;
import org.apache.camel.support.resume.Offsets;
import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.util.IOHelper;
+import org.apache.camel.util.StopWatch;
import org.apache.camel.util.StringHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -318,7 +319,15 @@ public class Splitter extends MulticastProcessor {
storedIndex = readCurrentWatermark();
}
if (storedIndex != null) {
- int skipTo = Integer.parseInt(storedIndex);
+ int skipTo;
+ try {
+ skipTo = Integer.parseInt(storedIndex);
+ } catch (NumberFormatException e) {
+ throw new RuntimeCamelException(
+ "Watermark value '" + storedIndex + "' under
key '" + watermarkKey
+ + "' is not a valid
integer",
+ e);
+ }
while (rawIterator.hasNext() && skipCount <= skipTo) {
rawIterator.next();
skipCount++;
@@ -549,6 +558,15 @@ public class Splitter extends MulticastProcessor {
this.watermarkExpression = watermarkExpression;
}
+ @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();
+ }
+ }
+
@Override
protected boolean shouldContinueOnFailure(Exchange subExchange, Exchange
original, int index) {
// only honor the tracker when THIS splitter has thresholds configured,
@@ -561,31 +579,43 @@ public class Splitter extends MulticastProcessor {
return super.shouldContinueOnFailure(subExchange, original, index);
}
- // record the failure
- tracker.recordFailure(index, subExchange.getException());
-
- // check if we've exceeded the max failed records
- if (maxFailedRecords > 0 && tracker.getFailureCount() >=
maxFailedRecords) {
+ // a deliberate .stop() or rollback in the child route is not a
failure —
+ // honor the stop/rollback semantics without inflating the failure
count
+ if (subExchange.isRouteStop() || subExchange.isRollbackOnly() ||
subExchange.isRollbackOnlyLast()) {
return false;
}
- // check if we've exceeded the error ratio threshold
- if (errorThreshold > 0) {
- double ratio = (double) tracker.getFailureCount() / (index + 1);
- if (ratio >= errorThreshold) {
+
+ // only record actual unhandled exceptions as failures —
error-handler-handled
+ // exceptions should not inflate the failure count
+ boolean hasException = subExchange.getException() != null;
+ if (hasException) {
+ tracker.recordFailure(index, subExchange.getException());
+
+ // check if we've exceeded the max failed records
+ if (maxFailedRecords > 0 && tracker.getFailureCount() >=
maxFailedRecords) {
return false;
}
- }
+ // check if we've exceeded the error ratio threshold
+ if (errorThreshold > 0) {
+ int processed = tracker.getProcessedItems();
+ if (processed > 0) {
+ double ratio = (double) tracker.getFailureCount() /
processed;
+ if (ratio >= errorThreshold) {
+ return false;
+ }
+ }
+ }
- // Continue processing — clear the exception from the sub-exchange so
that aggregation
- // proceeds normally. The failure is already recorded in the tracker
above, so the
- // SplitResult will still contain the failure details even though the
exception is cleared.
- subExchange.setException(null);
+ // continue processing — clear the exception so aggregation
proceeds normally
+ subExchange.setException(null);
+ }
return true;
}
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<>();
void recordFailure(int index, Exception exception) {
@@ -597,10 +627,18 @@ public class Splitter extends MulticastProcessor {
totalItems.incrementAndGet();
}
+ void incrementProcessedItems() {
+ processedItems.incrementAndGet();
+ }
+
int getTotalItems() {
return totalItems.get();
}
+ int getProcessedItems() {
+ return processedItems.get();
+ }
+
int getFailureCount() {
return failureCount.get();
}
@@ -722,8 +760,9 @@ public class Splitter extends MulticastProcessor {
}
boolean aborted = exchange.getException() != null;
- SplitResult result
- = new SplitResult(tracker.getTotalItems(),
tracker.getFailureCount(), tracker.getFailures(), aborted);
+ SplitResult result = new SplitResult(
+ tracker.getTotalItems(), tracker.getProcessedItems(),
+ tracker.getFailureCount(), tracker.getFailures(), aborted);
exchange.setProperty(ExchangePropertyKey.SPLIT_RESULT, result);
// remove internal tracker
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
index a9c8debbb09c..7bc663d7283b 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
@@ -212,6 +212,7 @@ public class AggregateProcessor extends BaseProcessorSupport
completedByStrategy.set(0);
completedByTimeout.set(0);
completedByPredicate.set(0);
+ completedByInterval.set(0);
completedByBatchConsumer.set(0);
completedByForce.set(0);
discarded.set(0);
@@ -404,9 +405,14 @@ public class AggregateProcessor extends
BaseProcessorSupport
long delay = optimisticLockRetryPolicy.getDelay(attempt);
if (delay > 0) {
int nextAttempt = attempt;
- getOptimisticLockingExecutorService().schedule(
- () -> doInOptimisticLock(exchange, key,
callback, nextAttempt, false), delay,
- TimeUnit.MILLISECONDS);
+ getOptimisticLockingExecutorService().schedule(() -> {
+ try {
+ doInOptimisticLock(exchange, key, callback,
nextAttempt, false);
+ } catch (Exception t) {
+ exchange.setException(t);
+ callback.done(false);
+ }
+ }, delay, TimeUnit.MILLISECONDS);
return false;
}
} else {
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterErrorThresholdTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterErrorThresholdTest.java
index 48da1eec0ded..c833f759fc54 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterErrorThresholdTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterErrorThresholdTest.java
@@ -100,8 +100,10 @@ class SplitterErrorThresholdTest extends
ContextTestSupport {
MockEndpoint mock = getMockEndpoint("mock:parallel-split");
mock.expectedMinimumMessageCount(0);
+ // 3 failures out of 5 items = 60% failure rate, exceeds 50% threshold
+ // regardless of parallel completion order
Exchange result = template.send("direct:parallel",
- e -> e.getIn().setBody(Arrays.asList("FAIL", "FAIL", "a", "b",
"c")));
+ e -> e.getIn().setBody(Arrays.asList("FAIL", "FAIL", "FAIL",
"a", "b")));
mock.assertIsSatisfied();
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterSplitResultTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterSplitResultTest.java
index d6603f6f3021..0fc517577011 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterSplitResultTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterSplitResultTest.java
@@ -136,8 +136,9 @@ class SplitterSplitResultTest extends ContextTestSupport {
@Test
void testSplitResultConstructorWithNullFailures() {
- SplitResult result = new SplitResult(5, 0, null, false);
+ SplitResult result = new SplitResult(5, 5, 0, null, false);
assertEquals(5, result.getTotalItems());
+ assertEquals(5, result.getProcessedItems());
assertEquals(0, result.getFailureCount());
assertEquals(5, result.getSuccessCount());
assertFalse(result.isAborted());
