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 cb22104c77fa CAMEL-24062: Fix Multicast EIP to honor
UseOriginalAggregationStrategy
cb22104c77fa is described below
commit cb22104c77fa13f56467e14778f021c3ac28925d
Author: Claus Ibsen <[email protected]>
AuthorDate: Wed Jul 15 09:10:08 2026 +0200
CAMEL-24062: Fix Multicast EIP to honor UseOriginalAggregationStrategy
Move the UseOriginalAggregationStrategy binding from Splitter and
RecipientListProcessor into MulticastProcessor (their shared base class),
so all three EIPs consistently bind the original exchange on the strategy.
Previously Multicast never called newInstance(exchange), making the strategy
silently ineffective in error scenarios where the aggregated result could
overwrite the original exchange body.
Closes #24697
Co-Authored-By: Claude Opus 4.6 <[email protected]>
---
.../apache/camel/processor/MulticastProcessor.java | 14 ++++++++++++++
.../camel/processor/RecipientListProcessor.java | 20 --------------------
.../java/org/apache/camel/processor/Splitter.java | 10 ----------
...icastUseOriginalPropagateExceptionCaughtTest.java | 2 ++
.../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 7 +++++++
5 files changed, 23 insertions(+), 30 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 419e62ae4e4b..71b2fbf6aa28 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
@@ -52,6 +52,8 @@ import org.apache.camel.Route;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.StreamCache;
import org.apache.camel.Traceable;
+import org.apache.camel.processor.aggregate.ShareUnitOfWorkAggregationStrategy;
+import org.apache.camel.processor.aggregate.UseOriginalAggregationStrategy;
import org.apache.camel.processor.errorhandler.ErrorHandlerSupport;
import org.apache.camel.spi.AsyncProcessorAwaitManager;
import org.apache.camel.spi.ErrorHandlerAware;
@@ -320,6 +322,18 @@ public class MulticastProcessor extends
BaseProcessorSupport
@Override
public boolean process(Exchange exchange, AsyncCallback callback) {
+ AggregationStrategy strategy = getAggregationStrategy();
+
+ // set original exchange if not already pre-configured
+ if (strategy instanceof UseOriginalAggregationStrategy original) {
+ // need to create a new private instance, as we can also have
concurrency issue so we cannot store state
+ AggregationStrategy clone = original.newInstance(exchange);
+ if (isShareUnitOfWork()) {
+ clone = new ShareUnitOfWorkAggregationStrategy(clone);
+ }
+ setAggregationStrategyOnExchange(exchange, clone);
+ }
+
if (synchronous) {
try {
// force synchronous processing using await manager
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/RecipientListProcessor.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/RecipientListProcessor.java
index f946becd4634..a67cee877818 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/RecipientListProcessor.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/RecipientListProcessor.java
@@ -26,7 +26,6 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import org.apache.camel.AggregationStrategy;
-import org.apache.camel.AsyncCallback;
import org.apache.camel.AsyncProducer;
import org.apache.camel.CamelContext;
import org.apache.camel.Endpoint;
@@ -38,8 +37,6 @@ import org.apache.camel.NoTypeConversionAvailableException;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import org.apache.camel.Route;
-import org.apache.camel.processor.aggregate.ShareUnitOfWorkAggregationStrategy;
-import org.apache.camel.processor.aggregate.UseOriginalAggregationStrategy;
import org.apache.camel.spi.NormalizedEndpointUri;
import org.apache.camel.spi.ProducerCache;
import org.apache.camel.support.AsyncProcessorConverterHelper;
@@ -191,23 +188,6 @@ public class RecipientListProcessor extends
MulticastProcessor {
this.ignoreInvalidEndpoints = ignoreInvalidEndpoints;
}
- @Override
- public boolean process(Exchange exchange, final AsyncCallback callback) {
- AggregationStrategy strategy = getAggregationStrategy();
-
- // set original exchange if not already pre-configured
- if (strategy instanceof UseOriginalAggregationStrategy original) {
- // need to create a new private instance, as we can also have
concurrency issue so we cannot store state
- AggregationStrategy clone = original.newInstance(exchange);
- if (isShareUnitOfWork()) {
- clone = new ShareUnitOfWorkAggregationStrategy(clone);
- }
- setAggregationStrategyOnExchange(exchange, clone);
- }
-
- return super.process(exchange, callback);
- }
-
@Override
protected Iterable<ProcessorExchangePair>
createProcessorExchangePairs(Exchange exchange)
throws Exception {
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 841e9f610459..2669a1652a5a 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
@@ -186,16 +186,6 @@ public class Splitter extends MulticastProcessor {
public boolean process(Exchange exchange, final AsyncCallback callback) {
AggregationStrategy strategy = getAggregationStrategy();
- // set original exchange if not already pre-configured
- if (strategy instanceof UseOriginalAggregationStrategy original) {
- // need to create a new private instance, as we can also have
concurrency issue so we cannot store state
- AggregationStrategy clone = original.newInstance(exchange);
- if (isShareUnitOfWork()) {
- clone = new ShareUnitOfWorkAggregationStrategy(clone);
- }
- setAggregationStrategyOnExchange(exchange, clone);
- }
-
// if no custom aggregation strategy is being used then fallback to
keep the original
// and propagate exceptions which is done by a per exchange specific
aggregation strategy
// to ensure it supports async routing
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/MulticastUseOriginalPropagateExceptionCaughtTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/MulticastUseOriginalPropagateExceptionCaughtTest.java
index 0cf9ed2b3892..4d8eb952a6c3 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/MulticastUseOriginalPropagateExceptionCaughtTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/MulticastUseOriginalPropagateExceptionCaughtTest.java
@@ -45,6 +45,8 @@ public class MulticastUseOriginalPropagateExceptionCaughtTest
extends ContextTes
assertEquals(1,
getMockEndpoint("mock:result").getReceivedExchanges().size());
Exchange exchange =
getMockEndpoint("mock:result").getReceivedExchanges().get(0);
+ // verify original body is preserved
+ assertEquals(body, exchange.getIn().getBody(String.class));
Exception exception = exchange.getProperty(Exchange.EXCEPTION_CAUGHT,
Exception.class);
assertNotNull(exception);
Throwable rootCause = exception;
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
index 64862ee17cf0..859c222605bc 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
@@ -416,6 +416,13 @@ The clustered route controller property
`camel.clustered.controller.initial-dela
it keeps its `String` type because it supports Camel time patterns (for
example `10 seconds`)
that `Duration` binding does not.
+=== camel-core - Multicast UseOriginalAggregationStrategy fix
+
+The Multicast EIP now correctly honors `UseOriginalAggregationStrategy`,
consistent with the Splitter
+and Recipient List EIPs. Previously, Multicast did not bind the original
exchange on the strategy,
+so the strategy was silently ineffective — especially in error scenarios where
the aggregated result
+could overwrite the original exchange body instead of preserving it.
+
=== camel-sql - PostgreSQL aggregation repositories fixed
`PostgresAggregationRepository` and `ClusteredPostgresAggregationRepository`
were broken since
