This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-4.18.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.18.x by this push:
new 8f4eb37bc5ca CAMEL-24062: Fix Multicast EIP to honor
UseOriginalAggregationStrategy
8f4eb37bc5ca is described below
commit 8f4eb37bc5caec3dae5eab482ead113a3d1c1a66
Author: Claus Ibsen <[email protected]>
AuthorDate: Wed Jul 15 09:56:51 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 #24699
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
