Skip to content

CAMEL-25005: camel-core - Saga EIP: do not lose the compensation of a step that joins while the in-memory saga ends #26865

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
allthingssecurity wants to merge 1 commit into
base: main
Choose a base branch
from
+156 −18
Open

CAMEL-25005: camel-core - Saga EIP: do not lose the compensation of a step that joins while the in-memory saga ends #26865

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
89 changes: 89 additions & 0 deletions core/camel-core/src/test/java/org/apache/camel/processor/SagaJoinDuringTimeoutTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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");
}
};
}
}
85 changes: 67 additions & 18 deletions core/camel-support/src/main/java/org/apache/camel/saga/InMemorySagaCoordinator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -60,6 +62,8 @@ private enum Status {
private final List<StepEnlistment> enlistments;
private final List<ScheduledFuture<?>> timeoutFutures;
private final AtomicReference<Status> 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");
Expand Down Expand Up @@ -102,28 +106,42 @@ public CompletableFuture<Void> 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<Void> 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<StepEnlistment> 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);
}

@Override
public CompletableFuture<Void> compensate(Exchange exchange) {
boolean doAction = currentStatus.compareAndSet(Status.RUNNING, Status.COMPENSATING);
List<StepEnlistment> 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);
Expand All @@ -144,11 +162,11 @@ public CompletableFuture<Void> compensate(Exchange exchange) {

@Override
public CompletableFuture<Void> complete(Exchange exchange) {
boolean doAction = currentStatus.compareAndSet(Status.RUNNING, Status.COMPLETING);
List<StepEnlistment> 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);
Expand All @@ -167,8 +185,29 @@ public CompletableFuture<Void> 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
* <tt>null</tt> 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<StepEnlistment> end(Status status) {
lock.lock();
try {
if (currentStatus.compareAndSet(Status.RUNNING, status)) {
return new ArrayList<>(enlistments);
}
return null;
} finally {
lock.unlock();
}
}

public CompletableFuture<Boolean> doCompensate(final Exchange exchange) {
return doFinalize(exchange, CamelSagaStep::getCompensation, "compensation")
return doCompensate(exchange, enlistments);
}

private CompletableFuture<Boolean> doCompensate(final Exchange exchange, List<StepEnlistment> 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);
Expand All @@ -179,7 +218,11 @@ public CompletableFuture<Boolean> doCompensate(final Exchange exchange) {
}

public CompletableFuture<Boolean> doComplete(final Exchange exchange) {
return doFinalize(exchange, CamelSagaStep::getCompletion, "completion")
return doComplete(exchange, enlistments);
}

private CompletableFuture<Boolean> doComplete(final Exchange exchange, List<StepEnlistment> 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);
Expand All @@ -192,8 +235,14 @@ public CompletableFuture<Boolean> doComplete(final Exchange exchange) {
public CompletableFuture<Boolean> doFinalize(
final Exchange exchange,
Function<CamelSagaStep, Optional<Endpoint>> endpointExtractor, String description) {
return doFinalize(exchange, enlistments, endpointExtractor, description);
}

private CompletableFuture<Boolean> doFinalize(
final Exchange exchange, List<StepEnlistment> steps,
Function<CamelSagaStep, Optional<Endpoint>> endpointExtractor, String description) {
CompletableFuture<Boolean> result = CompletableFuture.completedFuture(true);
for (StepEnlistment enlistment : reversed(enlistments)) {
for (StepEnlistment enlistment : reversed(steps)) {
Optional<Endpoint> endpoint = endpointExtractor.apply(enlistment.step);
if (endpoint.isPresent()) {
result = result.thenCompose(
Expand Down