From f63ba552f609814662bd320ae31e729b545778f4 Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Thu, 24 Sep 2026 22:11:02 +0530 Subject: [PATCH 1/2] CAMEL-XXXXX: camel-core - Saga EIP: keep the saga of an exchange in its copies Since CAMEL-23469 the saga id travels in the exchange's internal state, but the copy constructor of AbstractExchange, used by Exchange.copy(), did not copy it. The sub-exchanges of split, multicast, recipient list and wire tap are copies, so they only kept the saga through the Long-Running-Action header. Since CAMEL-24449 SagaProcessor reads that header only for saga services that support it, which the default InMemorySagaService does not. So a saga step reached through split, multicast, recipient list or wire tap no longer saw the saga: MANDATORY failed with "Exchange is not part of a saga", REQUIRED started and completed a saga per sub-exchange even when the parent saga was compensated, and SUPPORTS ran outside of any saga. The copy constructor now copies the saga id, so a copy belongs to the same saga as the exchange it was copied from, as it did through the header before. The internal state is only set by Camel, so a message still cannot choose its saga (CAMEL-24449). Regression from the combination of CAMEL-23469 and CAMEL-24449. Co-Authored-By: Claude Opus 5.5 --- .../camel/processor/SagaExchangeCopyTest.java | 89 +++++++++++++++++++ .../camel/support/AbstractExchange.java | 2 + 2 files changed, 91 insertions(+) create mode 100644 core/camel-core/src/test/java/org/apache/camel/processor/SagaExchangeCopyTest.java diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/SagaExchangeCopyTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/SagaExchangeCopyTest.java new file mode 100644 index 0000000000000..23450c294efb6 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/SagaExchangeCopyTest.java @@ -0,0 +1,89 @@ +/* + * 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 java.util.List; + +import org.apache.camel.CamelExecutionException; +import org.apache.camel.ContextTestSupport; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.model.SagaPropagation; +import org.apache.camel.saga.InMemorySagaService; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +/** + * The exchanges created by split and multicast are copies of the exchange of the saga, and belong to the same saga. + */ +class SagaExchangeCopyTest extends ContextTestSupport { + + @Test + void testSplitMandatoryStepsJoinSaga() throws Exception { + assertItemsCompensated("direct:split", 3); + } + + @Test + void testMulticastRequiredStepsJoinSaga() throws Exception { + assertItemsCompensated("direct:multicast", 2); + } + + private void assertItemsCompensated(String uri, int items) throws Exception { + getMockEndpoint("mock:compensate").expectedMessageCount(1); + getMockEndpoint("mock:item").expectedMessageCount(items); + getMockEndpoint("mock:compensate-item").expectedMessageCount(items); + getMockEndpoint("mock:complete-item").expectedMessageCount(0); + + CamelExecutionException ex + = assertThrows(CamelExecutionException.class, () -> template.sendBody(uri, List.of("a", "b", "c"))); + + MockEndpoint.assertIsSatisfied(context); + assertEquals("payment declined", ex.getCause().getMessage()); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() throws Exception { + context.addService(new InMemorySagaService()); + + from("direct:split") + .saga().compensation("mock:compensate") + .split(body()).to("direct:mandatory-item").end() + .throwException(new IllegalStateException("payment declined")); + + from("direct:multicast") + .saga().compensation("mock:compensate") + .multicast().to("direct:required-item", "direct:required-item").end() + .throwException(new IllegalStateException("payment declined")); + + from("direct:mandatory-item") + .saga().propagation(SagaPropagation.MANDATORY) + .compensation("mock:compensate-item").completion("mock:complete-item") + .to("mock:item"); + + from("direct:required-item") + .saga().propagation(SagaPropagation.REQUIRED) + .compensation("mock:compensate-item").completion("mock:complete-item") + .to("mock:item"); + } + }; + } +} diff --git a/core/camel-support/src/main/java/org/apache/camel/support/AbstractExchange.java b/core/camel-support/src/main/java/org/apache/camel/support/AbstractExchange.java index ce4bf980344b4..fdaa2c467fe64 100644 --- a/core/camel-support/src/main/java/org/apache/camel/support/AbstractExchange.java +++ b/core/camel-support/src/main/java/org/apache/camel/support/AbstractExchange.java @@ -143,6 +143,8 @@ protected AbstractExchange(AbstractExchange parent) { this.rollbackOnly = parent.rollbackOnly; this.rollbackOnlyLast = parent.rollbackOnlyLast; this.routeStop = parent.routeStop; + // a copy (such as a split, multicast or wire tap exchange) belongs to the same saga + this.sagaLongRunningAction = parent.sagaLongRunningAction; if (parent.hasVariables()) { if (this.variableRepository == null) { From ad865187885b9f1dedbe458dd143d8d8ae980ee5 Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Fri, 25 Sep 2026 08:39:06 +0530 Subject: [PATCH 2/2] CAMEL-25011: camel-core - Saga EIP: keep the saga of an exchange in its copies Since CAMEL-23469 the saga id travels in the exchange's internal state, but the copy constructor of AbstractExchange, used by Exchange.copy(), did not copy it. The sub-exchanges of split, multicast, recipient list and wire tap are copies, so they only kept the saga through the Long-Running-Action header. Since CAMEL-24449 SagaProcessor reads that header only for saga services that support it, which the default InMemorySagaService does not. So a saga step reached through split, multicast, recipient list or wire tap no longer saw the saga: MANDATORY failed with "Exchange is not part of a saga", REQUIRED started and completed a saga per sub-exchange even when the parent saga was compensated, and SUPPORTS ran outside of any saga. ExchangeHelper.copyResults also copies the saga id from a copy back to the original exchange, so after a multicast, recipient list, routing slip, failover or loop with copy, the original exchange lost its saga as well, and so did the seda consumer of the documented MANUAL completion example. The copy constructor now copies the saga id, so a copy belongs to the same saga as the exchange it was copied from, as it did through the header before. The internal state is only set by Camel, so a message still cannot choose its saga (CAMEL-24449). Regression from the combination of CAMEL-23469 and CAMEL-24449. Co-Authored-By: Claude Opus 5.5 --- .../apache/camel/processor/SagaExchangeCopyTest.java | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/SagaExchangeCopyTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/SagaExchangeCopyTest.java index 23450c294efb6..7a7de9ce67ac0 100644 --- a/core/camel-core/src/test/java/org/apache/camel/processor/SagaExchangeCopyTest.java +++ b/core/camel-core/src/test/java/org/apache/camel/processor/SagaExchangeCopyTest.java @@ -44,6 +44,12 @@ void testMulticastRequiredStepsJoinSaga() throws Exception { assertItemsCompensated("direct:multicast", 2); } + @Test + void testStepAfterMulticastJoinsSaga() throws Exception { + // the multicast copies the result of its last copy back to the exchange, which must keep the saga + assertItemsCompensated("direct:after-multicast", 1); + } + private void assertItemsCompensated(String uri, int items) throws Exception { getMockEndpoint("mock:compensate").expectedMessageCount(1); getMockEndpoint("mock:item").expectedMessageCount(items); @@ -74,6 +80,12 @@ public void configure() throws Exception { .multicast().to("direct:required-item", "direct:required-item").end() .throwException(new IllegalStateException("payment declined")); + from("direct:after-multicast") + .saga().compensation("mock:compensate") + .multicast().to("mock:a", "mock:b").end() + .to("direct:mandatory-item") + .throwException(new IllegalStateException("payment declined")); + from("direct:mandatory-item") .saga().propagation(SagaPropagation.MANDATORY) .compensation("mock:compensate-item").completion("mock:complete-item")