This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24062 in repository https://gitbox.apache.org/repos/asf/camel.git
commit e156a136b665e26a535e27623c7959f02b1110b5 Author: Claus Ibsen <[email protected]> AuthorDate: Tue Jul 14 22:16:46 2026 +0200 CAMEL-24062: Fix Multicast EIP to honor UseOriginalAggregationStrategy Moved the UseOriginalAggregationStrategy binding from Splitter and RecipientListProcessor into MulticastProcessor (their shared base), 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. Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[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 0454ac8a835a..a9b7c814d3ba 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 @@ -364,6 +364,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
