This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24145 in repository https://gitbox.apache.org/repos/asf/camel.git
commit 948dec61462a0bf6596f0b9c30f211a1bbc883ea Author: Claus Ibsen <[email protected]> AuthorDate: Fri Jul 17 10:25:06 2026 +0200 CAMEL-24145: camel-saga - InMemorySagaCoordinator compensate/complete now propagate finalization outcome Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../apache/camel/processor/SagaFailuresTest.java | 21 +++++++++++ .../apache/camel/saga/InMemorySagaCoordinator.java | 44 ++++++++++++++-------- .../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 17 +++++++++ 3 files changed, 66 insertions(+), 16 deletions(-) diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/SagaFailuresTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/SagaFailuresTest.java index 3ea66d96c136..67d8d20636be 100644 --- a/core/camel-core/src/test/java/org/apache/camel/processor/SagaFailuresTest.java +++ b/core/camel-core/src/test/java/org/apache/camel/processor/SagaFailuresTest.java @@ -19,11 +19,14 @@ package org.apache.camel.processor; import java.util.concurrent.atomic.AtomicInteger; import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; import org.apache.camel.builder.RouteBuilder; import org.apache.camel.component.mock.MockEndpoint; import org.apache.camel.saga.InMemorySagaService; import org.junit.jupiter.api.Test; +import static org.junit.jupiter.api.Assertions.assertNotNull; + public class SagaFailuresTest extends ContextTestSupport { private AtomicInteger maxFailures; @@ -86,6 +89,24 @@ public class SagaFailuresTest extends ContextTestSupport { end.assertIsSatisfied(); } + @Test + public void testCompletionFailurePropagates() throws Exception { + maxFailures = new AtomicInteger(3); + + Exchange result = template.send("direct:saga-complete", e -> e.getMessage().setBody("hello")); + + assertNotNull(result.getException(), "Completion failure should propagate to exchange"); + } + + @Test + public void testCompensationFailurePropagates() throws Exception { + maxFailures = new AtomicInteger(3); + + Exchange result = template.send("direct:saga-compensate", e -> e.getMessage().setBody("hello")); + + assertNotNull(result.getException(), "Compensation failure should propagate to exchange"); + } + @Override protected RouteBuilder createRouteBuilder() { diff --git a/core/camel-support/src/main/java/org/apache/camel/saga/InMemorySagaCoordinator.java b/core/camel-support/src/main/java/org/apache/camel/saga/InMemorySagaCoordinator.java index 4811325c9d88..85effefdb57b 100644 --- a/core/camel-support/src/main/java/org/apache/camel/saga/InMemorySagaCoordinator.java +++ b/core/camel-support/src/main/java/org/apache/camel/saga/InMemorySagaCoordinator.java @@ -123,14 +123,20 @@ public class InMemorySagaCoordinator implements CamelSagaCoordinator { boolean doAction = currentStatus.compareAndSet(Status.RUNNING, Status.COMPENSATING); if (doAction) { - doCompensate(exchange); - } else { - Status status = currentStatus.get(); - if (status != Status.COMPENSATING && status != Status.COMPENSATED) { - CompletableFuture<Void> res = new CompletableFuture<>(); - res.completeExceptionally(new IllegalStateException("Cannot compensate: status is " + status)); - return res; - } + return doCompensate(exchange).thenApply(res -> { + if (!res) { + throw new RuntimeCamelException( + "Unable to compensate all required steps of the saga " + sagaId); + } + return null; + }); + } + + Status status = currentStatus.get(); + if (status != Status.COMPENSATING && status != Status.COMPENSATED) { + CompletableFuture<Void> res = new CompletableFuture<>(); + res.completeExceptionally(new IllegalStateException("Cannot compensate: status is " + status)); + return res; } return CompletableFuture.completedFuture(null); @@ -141,14 +147,20 @@ public class InMemorySagaCoordinator implements CamelSagaCoordinator { boolean doAction = currentStatus.compareAndSet(Status.RUNNING, Status.COMPLETING); if (doAction) { - doComplete(exchange); - } else { - Status status = currentStatus.get(); - if (status != Status.COMPLETING && status != Status.COMPLETED) { - CompletableFuture<Void> res = new CompletableFuture<>(); - res.completeExceptionally(new IllegalStateException("Cannot complete: status is " + status)); - return res; - } + return doComplete(exchange).thenApply(res -> { + if (!res) { + throw new RuntimeCamelException( + "Unable to complete all required steps of the saga " + sagaId); + } + return null; + }); + } + + Status status = currentStatus.get(); + if (status != Status.COMPLETING && status != Status.COMPLETED) { + CompletableFuture<Void> res = new CompletableFuture<>(); + res.completeExceptionally(new IllegalStateException("Cannot complete: status is " + status)); + return res; } return CompletableFuture.completedFuture(null); diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc index c98427b5f5f1..add42ac4bf48 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc @@ -553,6 +553,23 @@ stylesheet-driven external fetches. Users of Saxon Professional or Enterprise editions who rely on Java extension functions called from XSLT stylesheets must now explicitly set `secureProcessing=false` on the endpoint. +=== camel-core - InMemorySagaCoordinator now propagates finalization outcome + +The `InMemorySagaCoordinator` used for the Saga EIP (when no external LRA coordinator is configured) +previously returned an already-completed future from `compensate()` and `complete()`, meaning the +exchange finished before any compensation or completion endpoints had been invoked. Finalization +failures were only logged as warnings and never propagated to the caller. + +The coordinator now returns the actual finalization future, so the exchange waits for the +compensation or completion callbacks to finish (including retries). If all retry attempts fail, +the failure is propagated as an exception on the exchange, consistent with the `camel-lra` +coordinator behavior. + +This is a behavior change: routes that previously completed immediately regardless of +compensation/completion outcome will now wait for finalization and may see exceptions +that were previously swallowed. If your route relies on fire-and-forget saga finalization, +consider using `MANUAL` completion mode instead. + === camel-xslt / camel-xslt-saxon - transformerFactoryConfigurationStrategy now honored The `transformerFactoryConfigurationStrategy` option is now applied on all factory creation paths.
