This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch backport/24697-to-camel-4.18.x in repository https://gitbox.apache.org/repos/asf/camel.git
commit f6c9c7de44f73ed7f130227f1e1a74b081e8731f 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_18.adoc | 9 +++++++++ 5 files changed, 25 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 1066873a6331..f21b73704c20 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; @@ -308,6 +310,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 4620ecf945f4..d746813b1d36 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 @@ -102,16 +102,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_18.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_18.adoc index 88002166e0b9..1aea803fd6f9 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_18.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_18.adoc @@ -13,6 +13,15 @@ See the xref:camel-upgrade-recipes-tool.adoc[documentation] page for details. == Upgrading from 4.18.3 to 4.18.4 +=== camel-core - Multicast EIP honors UseOriginalAggregationStrategy + +The `UseOriginalAggregationStrategy` binding has been moved from `Splitter` and `RecipientListProcessor` +into `MulticastProcessor` (their shared base class), so all three EIPs (Multicast, Splitter, +RecipientList) now consistently bind the original exchange on the strategy. Previously, Multicast +never called `newInstance(exchange)`, which made the strategy silently ineffective — especially in +error scenarios where the aggregated result could overwrite the original exchange body instead of +preserving it. + === camel-azure-servicebus - Camel-managed message lock renewal When consuming from Azure Service Bus in `PEEK_LOCK` mode with `maxAutoLockRenewDuration > 0`, Camel now
