From 333b5326e4ae09699112aeb4b6d98a0aa08e9149 Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Thu, 24 Sep 2026 22:05:17 +0530 Subject: [PATCH] CAMEL-25005: camel-core - Saga EIP: do not lose the compensation of a step that joins while the in-memory saga ends InMemorySagaCoordinator.beginStep checked that the saga was RUNNING, evaluated the step's options, and then added the step to the enlistments, without any lock. complete(), compensate() and the timeout task changed the status with a compareAndSet and then finalized a snapshot of the enlistments. A step that passed the check before the saga ended, but enlisted after the snapshot, got a successful beginStep, so its action ran, but it was never passed to its compensation or completion endpoint. The window includes the option expressions, and the saga timeout makes it reachable with synchronous routes only. beginStep now checks the status again and enlists the step under a lock of the coordinator, and complete(), compensate() and the timeout task change the status and take the snapshot they finalize under the same lock. A step that loses the race fails with "Cannot begin: status is ...", as it does when it arrives a moment later. Co-Authored-By: Claude Opus 5.5 --- .../processor/SagaJoinDuringTimeoutTest.java | 89 +++++++++++++++++++ .../camel/saga/InMemorySagaCoordinator.java | 85 ++++++++++++++---- 2 files changed, 156 insertions(+), 18 deletions(-) create mode 100644 core/camel-core/src/test/java/org/apache/camel/processor/SagaJoinDuringTimeoutTest.java diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/SagaJoinDuringTimeoutTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/SagaJoinDuringTimeoutTest.java new file mode 100644 index 0000000000000..9d113d5e835e4 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/SagaJoinDuringTimeoutTest.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.concurrent.TimeUnit; + +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.SagaPropagation; +import org.apache.camel.saga.InMemorySagaService; +import org.apache.camel.support.ExpressionAdapter; +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; + +/** + * A step that joins a saga while the saga times out must either be compensated with the saga, or fail to begin. It must + * not run without ever being compensated. + */ +class SagaJoinDuringTimeoutTest extends ContextTestSupport { + + @Test + void testStepJoiningWhileSagaTimesOut() throws Exception { + MockEndpoint compensate = getMockEndpoint("mock:compensate"); + compensate.expectedMessageCount(1); + + MockEndpoint payment = getMockEndpoint("mock:payment"); + payment.expectedMessageCount(0); + + MockEndpoint compensatePayment = getMockEndpoint("mock:compensate-payment"); + compensatePayment.expectedMessageCount(0); + + CamelExecutionException ex + = assertThrows(CamelExecutionException.class, () -> template.sendBody("direct:order", "Hello")); + + MockEndpoint.assertIsSatisfied(context); + assertInstanceOf(IllegalStateException.class, ex.getCause()); + assertTrue(ex.getCause().getMessage().startsWith("Cannot begin: status is COMPENSAT"), ex.getCause().getMessage()); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() throws Exception { + context.addService(new InMemorySagaService()); + + from("direct:order") + .saga().timeout(100, TimeUnit.MILLISECONDS).compensation("mock:compensate") + .to("direct:payment"); + + from("direct:payment") + .saga().propagation(SagaPropagation.MANDATORY) + .option("orderId", new ExpressionAdapter() { + @Override + public Object evaluate(Exchange exchange) { + // the options are evaluated when the step joins the saga, and this takes until the + // saga has timed out and its compensation is running + await().atMost(5, TimeUnit.SECONDS) + .until(() -> getMockEndpoint("mock:compensate").getReceivedCounter() > 0); + return "order-1"; + } + }) + .compensation("mock:compensate-payment") + .to("mock:payment"); + } + }; + } +} 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 7019d7fa5b20f..a91fed79609e4 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 @@ -27,6 +27,8 @@ import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import java.util.function.Function; import org.apache.camel.CamelContext; @@ -60,6 +62,8 @@ private enum Status { private final List enlistments; private final List> timeoutFutures; private final AtomicReference currentStatus; + // makes the status check and the enlistment of a step atomic with the change of status that ends the saga + private final Lock lock = new ReentrantLock(); public InMemorySagaCoordinator(CamelContext camelContext, InMemorySagaService sagaService, String sagaId) { this.camelContext = ObjectHelper.notNull(camelContext, "camelContext"); @@ -102,16 +106,30 @@ public CompletableFuture beginStep(Exchange exchange, CamelSagaStep step) } } } - this.enlistments.add(new StepEnlistment(step, values)); - if (step.getTimeoutInMilliseconds().isPresent()) { - ScheduledFuture timeoutFuture = sagaService.getExecutorService().schedule(() -> { - boolean doAction = currentStatus.compareAndSet(Status.RUNNING, Status.COMPENSATING); - if (doAction) { - doCompensate(exchange); - } - }, step.getTimeoutInMilliseconds().get(), TimeUnit.MILLISECONDS); - timeoutFutures.add(timeoutFuture); + lock.lock(); + try { + // check again, as the saga may have been completed, compensated or timed out while the options were + // evaluated, and then the step would not be finalized + status = currentStatus.get(); + if (status != Status.RUNNING) { + CompletableFuture res = new CompletableFuture<>(); + res.completeExceptionally(new IllegalStateException("Cannot begin: status is " + status)); + return res; + } + this.enlistments.add(new StepEnlistment(step, values)); + + if (step.getTimeoutInMilliseconds().isPresent()) { + ScheduledFuture timeoutFuture = sagaService.getExecutorService().schedule(() -> { + List steps = end(Status.COMPENSATING); + if (steps != null) { + doCompensate(exchange, steps); + } + }, step.getTimeoutInMilliseconds().get(), TimeUnit.MILLISECONDS); + timeoutFutures.add(timeoutFuture); + } + } finally { + lock.unlock(); } return CompletableFuture.completedFuture(null); @@ -119,11 +137,11 @@ public CompletableFuture beginStep(Exchange exchange, CamelSagaStep step) @Override public CompletableFuture compensate(Exchange exchange) { - boolean doAction = currentStatus.compareAndSet(Status.RUNNING, Status.COMPENSATING); + List steps = end(Status.COMPENSATING); - if (doAction) { + if (steps != null) { cancelTimeouts(); - return doCompensate(exchange).thenApply(res -> { + return doCompensate(exchange, steps).thenApply(res -> { if (!res) { throw new RuntimeCamelException( "Unable to compensate all required steps of the saga " + sagaId); @@ -144,11 +162,11 @@ public CompletableFuture compensate(Exchange exchange) { @Override public CompletableFuture complete(Exchange exchange) { - boolean doAction = currentStatus.compareAndSet(Status.RUNNING, Status.COMPLETING); + List steps = end(Status.COMPLETING); - if (doAction) { + if (steps != null) { cancelTimeouts(); - return doComplete(exchange).thenApply(res -> { + return doComplete(exchange, steps).thenApply(res -> { if (!res) { throw new RuntimeCamelException( "Unable to complete all required steps of the saga " + sagaId); @@ -167,8 +185,29 @@ public CompletableFuture complete(Exchange exchange) { return CompletableFuture.completedFuture(null); } + /** + * Changes the status from RUNNING to the given status, and returns the steps enlisted at that time, or + * null if the saga is not running. This holds the lock that beginStep holds while it checks the status and + * enlists a step, so every step that began is in the returned list, and is finalized. + */ + private List end(Status status) { + lock.lock(); + try { + if (currentStatus.compareAndSet(Status.RUNNING, status)) { + return new ArrayList<>(enlistments); + } + return null; + } finally { + lock.unlock(); + } + } + public CompletableFuture doCompensate(final Exchange exchange) { - return doFinalize(exchange, CamelSagaStep::getCompensation, "compensation") + return doCompensate(exchange, enlistments); + } + + private CompletableFuture doCompensate(final Exchange exchange, List steps) { + return doFinalize(exchange, steps, CamelSagaStep::getCompensation, "compensation") .whenComplete((res, ex) -> { if (ex != null || !Boolean.TRUE.equals(res)) { LOG.warn("Saga {} compensation did not fully succeed — manual intervention may be needed", sagaId); @@ -179,7 +218,11 @@ public CompletableFuture doCompensate(final Exchange exchange) { } public CompletableFuture doComplete(final Exchange exchange) { - return doFinalize(exchange, CamelSagaStep::getCompletion, "completion") + return doComplete(exchange, enlistments); + } + + private CompletableFuture doComplete(final Exchange exchange, List steps) { + return doFinalize(exchange, steps, CamelSagaStep::getCompletion, "completion") .whenComplete((res, ex) -> { if (ex != null || !Boolean.TRUE.equals(res)) { LOG.warn("Saga {} completion did not fully succeed — manual intervention may be needed", sagaId); @@ -192,8 +235,14 @@ public CompletableFuture doComplete(final Exchange exchange) { public CompletableFuture doFinalize( final Exchange exchange, Function> endpointExtractor, String description) { + return doFinalize(exchange, enlistments, endpointExtractor, description); + } + + private CompletableFuture doFinalize( + final Exchange exchange, List steps, + Function> endpointExtractor, String description) { CompletableFuture result = CompletableFuture.completedFuture(true); - for (StepEnlistment enlistment : reversed(enlistments)) { + for (StepEnlistment enlistment : reversed(steps)) { Optional endpoint = endpointExtractor.apply(enlistment.step); if (endpoint.isPresent()) { result = result.thenCompose(