From 32242bcb029921a09d38af915111bb74374f5129 Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Thu, 24 Sep 2026 21:57:47 +0530 Subject: [PATCH] CAMEL-25006: camel-core - Saga EIP: fail a REQUIRED or SUPPORTS step of a saga that has already ended Since CAMEL-24144 (4.22.0) InMemorySagaService removes a saga once it is completed or compensated, and getSaga(id) then returns null. The saga processors treat null as "the exchange is not in a saga". So after a saga timed out and was compensated, a REQUIRED step on an exchange of that saga started a new saga of its own and completed it, and a SUPPORTS step ran outside of any saga. For example, a payment was taken and confirmed although the order it belonged to was compensated. Before CAMEL-24144, and still while the saga is being compensated, such a step failed with "Cannot begin: status is ...". RequiredSagaProcessor and SupportsSagaProcessor now fail with IllegalStateException when the exchange carries the id of a saga that the saga service no longer knows. The saga is still removed from the service, so this keeps the memory fix of CAMEL-24144. MANDATORY already fails, and REQUIRES_NEW, NOT_SUPPORTED and NEVER are unchanged. Regression from CAMEL-24144. Co-Authored-By: Claude Opus 5.5 --- .../processor/saga/RequiredSagaProcessor.java | 1 + .../camel/processor/saga/SagaProcessor.java | 33 ++++++++-- .../processor/saga/SupportsSagaProcessor.java | 1 + .../camel/processor/SagaTimeoutTest.java | 61 ++++++++++++++++++- .../pages/camel-4x-upgrade-guide-4_23.adoc | 15 +++++ 5 files changed, 105 insertions(+), 6 deletions(-) diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/RequiredSagaProcessor.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/RequiredSagaProcessor.java index f3f14cd7e7ae0..eb931fd3d2561 100644 --- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/RequiredSagaProcessor.java +++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/RequiredSagaProcessor.java @@ -39,6 +39,7 @@ public RequiredSagaProcessor(Processor childProcessor, CamelSagaService sagaServ public boolean process(Exchange exchange, AsyncCallback callback) { getCurrentSagaCoordinator(exchange) .whenComplete((existingCoordinator, ex) -> ifNotException(ex, exchange, callback, () -> { + checkSagaIsActive(exchange, existingCoordinator); CompletableFuture coordinatorFuture; final boolean inheritedCoordinator; if (existingCoordinator != null) { diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/SagaProcessor.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/SagaProcessor.java index cd14c5e97e227..efcb288144e9e 100644 --- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/SagaProcessor.java +++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/SagaProcessor.java @@ -53,6 +53,33 @@ protected SagaProcessor(Processor childProcessor, CamelSagaService sagaService, } protected CompletableFuture getCurrentSagaCoordinator(Exchange exchange) { + String currentSaga = getCurrentSagaId(exchange); + if (currentSaga != null) { + return sagaService.getSaga(currentSaga); + } + + return CompletableFuture.completedFuture(null); + } + + /** + * Checks that the saga of the exchange, if any, is still known by the saga service. The in-memory saga service + * removes a saga once it is completed or compensated (for example after a timeout), and then a step of that saga + * must fail, instead of starting a new saga or running outside of a saga. + * + * @param exchange the exchange + * @param coordinator the coordinator of the saga found for the exchange, or null if none + * @throws IllegalStateException if the exchange belongs to a saga that is no longer active + */ + protected void checkSagaIsActive(Exchange exchange, CamelSagaCoordinator coordinator) { + if (coordinator == null) { + String currentSaga = getCurrentSagaId(exchange); + if (currentSaga != null) { + throw new IllegalStateException("Cannot begin: saga " + currentSaga + " is not active or not known"); + } + } + } + + private String getCurrentSagaId(Exchange exchange) { // try internal state first (survives removeHeaders("*")) String currentSaga = exchange.getExchangeExtension().getSagaLongRunningAction(); if (currentSaga == null && sagaService.isLongRunningActionHeaderSupported()) { @@ -62,11 +89,7 @@ protected CompletableFuture getCurrentSagaCoordinator(Exch // message pick which saga its exchange joins. currentSaga = exchange.getIn().getHeader(Exchange.SAGA_LONG_RUNNING_ACTION, String.class); } - if (currentSaga != null) { - return sagaService.getSaga(currentSaga); - } - - return CompletableFuture.completedFuture(null); + return currentSaga; } protected void setCurrentSagaCoordinator(Exchange exchange, CamelSagaCoordinator coordinator) { diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/SupportsSagaProcessor.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/SupportsSagaProcessor.java index 70b8304bdc120..8e7916fb982d3 100644 --- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/SupportsSagaProcessor.java +++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/SupportsSagaProcessor.java @@ -38,6 +38,7 @@ public SupportsSagaProcessor(Processor childProcessor, CamelSagaService sagaServ @Override public boolean process(Exchange exchange, AsyncCallback callback) { getCurrentSagaCoordinator(exchange).whenComplete((coordinator, ex) -> ifNotException(ex, exchange, callback, () -> { + checkSagaIsActive(exchange, coordinator); if (coordinator != null) { coordinator.beginStep(exchange, step) .whenComplete((done, ex2) -> ifNotException(ex2, exchange, callback, () -> { diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/SagaTimeoutTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/SagaTimeoutTest.java index 87eebbef1cd0c..757f88dcb3926 100644 --- a/core/camel-core/src/test/java/org/apache/camel/processor/SagaTimeoutTest.java +++ b/core/camel-core/src/test/java/org/apache/camel/processor/SagaTimeoutTest.java @@ -21,6 +21,7 @@ import org.apache.camel.CamelExecutionException; 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.model.SagaCompletionMode; @@ -28,11 +29,15 @@ import org.apache.camel.saga.InMemorySagaService; import org.junit.jupiter.api.Test; +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; public class SagaTimeoutTest extends ContextTestSupport { + private InMemorySagaService sagaService; + @Test public void testTimeoutCalledCorrectly() throws Exception { MockEndpoint compensate = getMockEndpoint("mock:compensate"); @@ -94,12 +99,46 @@ public void testTimeoutMultiParticipants() throws Exception { compensate.assertIsSatisfied(); } + @Test + void testRequiredStepAfterTimeoutDoesNotStartNewSaga() throws Exception { + assertStepAfterTimeoutFails("direct:saga-timeout-required"); + } + + @Test + void testSupportsStepAfterTimeoutDoesNotRunOutsideSaga() throws Exception { + assertStepAfterTimeoutFails("direct:saga-timeout-supports"); + } + + private void assertStepAfterTimeoutFails(String uri) throws Exception { + MockEndpoint compensate = getMockEndpoint("mock:compensate"); + compensate.expectedMessageCount(1); + + MockEndpoint payment = getMockEndpoint("mock:payment"); + payment.expectedMessageCount(0); + + MockEndpoint complete = getMockEndpoint("mock:complete"); + complete.expectedMessageCount(0); + + CamelExecutionException ex = assertThrows(CamelExecutionException.class, () -> template.sendBody(uri, "Hello")); + + MockEndpoint.assertIsSatisfied(context); + assertInstanceOf(IllegalStateException.class, ex.getCause()); + assertTrue(ex.getCause().getMessage().endsWith("is not active or not known"), ex.getCause().getMessage()); + } + + private void awaitSagaEnded(Exchange exchange) { + // a slow call outlasts the saga timeout: the saga is compensated and removed from the saga service + String sagaId = exchange.getExchangeExtension().getSagaLongRunningAction(); + await().atMost(5, TimeUnit.SECONDS).until(() -> sagaService.getSaga(sagaId).get() == null); + } + @Override protected RouteBuilder createRouteBuilder() { return new RouteBuilder() { @Override public void configure() throws Exception { - context.addService(new InMemorySagaService()); + sagaService = new InMemorySagaService(); + context.addService(sagaService); from("direct:saga").saga().timeout(100, TimeUnit.MILLISECONDS).option("id", constant("myid")) .completionMode(SagaCompletionMode.MANUAL) @@ -130,6 +169,26 @@ public void configure() throws Exception { .propagation(SagaPropagation.MANDATORY).timeout(500, TimeUnit.MILLISECONDS) .compensation("mock:compensate").completion("mock:complete") .to("mock:end"); + + from("direct:saga-timeout-required") + .saga().timeout(100, TimeUnit.MILLISECONDS).compensation("mock:compensate") + .process(SagaTimeoutTest.this::awaitSagaEnded) + .to("direct:payment-required"); + + from("direct:payment-required") + .saga().propagation(SagaPropagation.REQUIRED) + .compensation("mock:compensate-payment").completion("mock:complete") + .to("mock:payment"); + + from("direct:saga-timeout-supports") + .saga().timeout(100, TimeUnit.MILLISECONDS).compensation("mock:compensate") + .process(SagaTimeoutTest.this::awaitSagaEnded) + .to("direct:payment-supports"); + + from("direct:payment-supports") + .saga().propagation(SagaPropagation.SUPPORTS) + .compensation("mock:compensate-payment") + .to("mock:payment"); } }; } diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc index f70b258eedc10..0df43c6bad915 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc @@ -1899,6 +1899,21 @@ A custom `CamelSagaService` that relies on the header to join sagas started by a override the new method. Everything else is unaffected: the header is still set on the exchange, and routes reading it continue to work. +=== camel-core - a saga step of a saga that has already ended fails + +Since 4.22, the `InMemorySagaService` removes a saga once it is completed or compensated, for example after +its timeout, or after `saga:complete` for a `MANUAL` saga. A saga step with propagation `REQUIRED` on an exchange +of such a saga then started a new saga of its own and completed it, and a step with propagation `SUPPORTS` ran +outside of any saga, although the saga of the exchange had already ended. + +Such a step now fails with `IllegalStateException` (`Cannot begin: saga is not active or not known`), as it +did before 4.22 and as it does while the saga is still being completed or compensated. A step with propagation +`MANDATORY` fails as before. + +The check applies to any `CamelSagaService` that returns no coordinator for the saga id of the exchange. For +example, a custom saga service that reads the `Long-Running-Action` header, and receives the id of a saga it does +not know, now fails such a step instead of starting a new saga. + === camel-microprofile-health The `error.stacktrace` entry of a failed health check is now only included in the response when