This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24140 in repository https://gitbox.apache.org/repos/asf/camel.git
commit 05b2c0f99bcdd3ec1eeaae59fdb35f5e71c775cf Author: Claus Ibsen <[email protected]> AuthorDate: Fri Jul 17 08:01:40 2026 +0200 CAMEL-24140: Fix UseOriginalAggregationStrategy not honored when shareUnitOfWork enabled Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../apache/camel/processor/MulticastProcessor.java | 3 + .../org/apache/camel/reifier/MulticastReifier.java | 6 -- .../apache/camel/reifier/RecipientListReifier.java | 6 -- .../org/apache/camel/reifier/SplitReifier.java | 5 -- .../MulticastUseOriginalShareUnitOfWorkTest.java | 86 ++++++++++++++++++++++ 5 files changed, 89 insertions(+), 17 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 5a970717ccca..a706092a6816 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 @@ -332,6 +332,9 @@ public class MulticastProcessor extends BaseProcessorSupport clone = new ShareUnitOfWorkAggregationStrategy(clone); } setAggregationStrategyOnExchange(exchange, clone); + } else if (isShareUnitOfWork() && strategy != null + && !(strategy instanceof ShareUnitOfWorkAggregationStrategy)) { + setAggregationStrategyOnExchange(exchange, new ShareUnitOfWorkAggregationStrategy(strategy)); } if (synchronous) { diff --git a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/MulticastReifier.java b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/MulticastReifier.java index b1ed396ff781..a50c7c1791eb 100644 --- a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/MulticastReifier.java +++ b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/MulticastReifier.java @@ -31,7 +31,6 @@ import org.apache.camel.model.ProcessorDefinition; import org.apache.camel.processor.MulticastProcessor; import org.apache.camel.processor.aggregate.AggregationStrategyBeanAdapter; import org.apache.camel.processor.aggregate.AggregationStrategyBiFunctionAdapter; -import org.apache.camel.processor.aggregate.ShareUnitOfWorkAggregationStrategy; import org.apache.camel.processor.aggregate.UseLatestAggregationStrategy; public class MulticastReifier extends ProcessorReifier<MulticastDefinition> { @@ -128,11 +127,6 @@ public class MulticastReifier extends ProcessorReifier<MulticastDefinition> { } CamelContextAware.trySetCamelContext(strategy, camelContext); - if (parseBoolean(definition.getShareUnitOfWork(), false)) { - // wrap strategy in share unit of work - strategy = new ShareUnitOfWorkAggregationStrategy(strategy); - } - return strategy; } diff --git a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/RecipientListReifier.java b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/RecipientListReifier.java index 992e731f35d4..6f7a4db173c3 100644 --- a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/RecipientListReifier.java +++ b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/RecipientListReifier.java @@ -32,7 +32,6 @@ import org.apache.camel.processor.EvaluateExpressionProcessor; import org.apache.camel.processor.RecipientList; import org.apache.camel.processor.aggregate.AggregationStrategyBeanAdapter; import org.apache.camel.processor.aggregate.AggregationStrategyBiFunctionAdapter; -import org.apache.camel.processor.aggregate.ShareUnitOfWorkAggregationStrategy; import org.apache.camel.processor.aggregate.UseLatestAggregationStrategy; public class RecipientListReifier extends ProcessorReifier<RecipientListDefinition<?>> { @@ -154,11 +153,6 @@ public class RecipientListReifier extends ProcessorReifier<RecipientListDefiniti } CamelContextAware.trySetCamelContext(strategy, camelContext); - if (parseBoolean(definition.getShareUnitOfWork(), false)) { - // wrap strategy in share unit of work - strategy = new ShareUnitOfWorkAggregationStrategy(strategy); - } - return strategy; } diff --git a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SplitReifier.java b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SplitReifier.java index 19b58ffec9f9..7c6e2c3db63c 100644 --- a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SplitReifier.java +++ b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SplitReifier.java @@ -29,7 +29,6 @@ import org.apache.camel.model.SplitDefinition; import org.apache.camel.processor.Splitter; import org.apache.camel.processor.aggregate.AggregationStrategyBeanAdapter; import org.apache.camel.processor.aggregate.AggregationStrategyBiFunctionAdapter; -import org.apache.camel.processor.aggregate.ShareUnitOfWorkAggregationStrategy; import org.apache.camel.resume.ResumeStrategy; public class SplitReifier extends ExpressionReifier<SplitDefinition> { @@ -182,10 +181,6 @@ public class SplitReifier extends ExpressionReifier<SplitDefinition> { } CamelContextAware.trySetCamelContext(strategy, camelContext); - if (strategy != null && parseBoolean(definition.getShareUnitOfWork(), false)) { - // wrap strategy in share unit of work - strategy = new ShareUnitOfWorkAggregationStrategy(strategy); - } return strategy; } diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/MulticastUseOriginalShareUnitOfWorkTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/MulticastUseOriginalShareUnitOfWorkTest.java new file mode 100644 index 000000000000..1540733009f2 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/MulticastUseOriginalShareUnitOfWorkTest.java @@ -0,0 +1,86 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.processor; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.processor.aggregate.UseOriginalAggregationStrategy; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class MulticastUseOriginalShareUnitOfWorkTest extends ContextTestSupport { + + @Test + public void testWithoutShareUnitOfWork() throws Exception { + getMockEndpoint("mock:a").expectedMessageCount(1); + getMockEndpoint("mock:b").expectedMessageCount(1); + getMockEndpoint("mock:result").expectedMessageCount(1); + + template.sendBody("direct:noShare", "original"); + + assertMockEndpointsSatisfied(); + + Exchange out = getMockEndpoint("mock:result").getReceivedExchanges().get(0); + assertEquals("original", out.getIn().getBody(String.class)); + } + + @Test + public void testWithShareUnitOfWork() throws Exception { + getMockEndpoint("mock:a").expectedMessageCount(1); + getMockEndpoint("mock:b").expectedMessageCount(1); + getMockEndpoint("mock:result").expectedMessageCount(1); + + template.sendBody("direct:share", "original"); + + assertMockEndpointsSatisfied(); + + Exchange out = getMockEndpoint("mock:result").getReceivedExchanges().get(0); + assertEquals("original", out.getIn().getBody(String.class)); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:noShare") + .multicast(new UseOriginalAggregationStrategy()) + .to("direct:a") + .to("direct:b") + .end() + .to("mock:result"); + + from("direct:share") + .multicast(new UseOriginalAggregationStrategy()).shareUnitOfWork() + .to("direct:a") + .to("direct:b") + .end() + .to("mock:result"); + + from("direct:a") + .setBody(constant("A")) + .to("mock:a"); + + from("direct:b") + .setBody(constant("B")) + .to("mock:b"); + } + }; + } +}
