CAMEL-25006: camel-core - Saga EIP: fail a REQUIRED or SUPPORTS step of a saga that has already ended

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<CamelSagaCoordinator> coordinatorFuture;
                     final boolean inheritedCoordinator;
                     if (existingCoordinator != null) {

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<CamelSagaCoordinator> 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 <tt>null</tt> 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<CamelSagaCoordinator> 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) {

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, () -> {

core/camel-core/src/test/java/org/apache/camel/processor/SagaTimeoutTest.java

@@ -21,18 +21,23 @@
 
 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;
 import org.apache.camel.model.SagaPropagation;
 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");
             }
         };
     }

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 <id> 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