This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 83c20f5550d3 CAMEL-25011: camel-core - Saga EIP: keep the saga of an
exchange in its copies (#26869)
83c20f5550d3 is described below
commit 83c20f5550d3bdc312fc1d3a42d397f3850e67c2
Author: allthingssecurity <[email protected]>
AuthorDate: Mon Sep 28 12:27:31 2026 +0530
CAMEL-25011: camel-core - Saga EIP: keep the saga of an exchange in its
copies (#26869)
The copy constructor of AbstractExchange now copies the saga id, so a
copy (split, multicast, recipient list, wire tap, seda, ...) belongs to the
same saga as the exchange it was copied from, and ExchangeHelper.copyResults
no longer clears it on the original exchange.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/processor/SagaExchangeCopyTest.java | 101 +++++++++++++++++++++
.../org/apache/camel/support/AbstractExchange.java | 2 +
2 files changed, 103 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
new file mode 100644
index 000000000000..7a7de9ce67ac
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/SagaExchangeCopyTest.java
@@ -0,0 +1,101 @@
+/*
+ * 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);
+ }
+
+ @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);
+ 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: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")
+ .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 ce4bf980344b..fdaa2c467fe6 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 @@ abstract class AbstractExchange implements Exchange,
ExchangeExtension {
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) {
