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(