This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new b7cfc0f148a2 CAMEL-24148: Fix saga option values overwritten on
repeated step enlistment
b7cfc0f148a2 is described below
commit b7cfc0f148a2b0cdd9d480512d0a9d051a2be305
Author: Claus Ibsen <[email protected]>
AuthorDate: Fri Jul 17 14:09:26 2026 +0200
CAMEL-24148: Fix saga option values overwritten on repeated step enlistment
Co-Authored-By: Claude Opus 4.6 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
Co-authored-by: Claude Opus 4.6 <[email protected]>
---
.../apache/camel/processor/SagaOptionsTest.java | 40 ++++++++++++++++
.../apache/camel/saga/InMemorySagaCoordinator.java | 53 ++++++++++++----------
2 files changed, 68 insertions(+), 25 deletions(-)
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/SagaOptionsTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/SagaOptionsTest.java
index aa7900bec82a..676328b27305 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/SagaOptionsTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/SagaOptionsTest.java
@@ -16,13 +16,18 @@
*/
package org.apache.camel.processor;
+import java.util.List;
+import java.util.stream.Collectors;
+
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.junit.jupiter.api.Test;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.fail;
public class SagaOptionsTest extends ContextTestSupport {
@@ -60,6 +65,27 @@ public class SagaOptionsTest extends ContextTestSupport {
compensate.assertIsSatisfied();
}
+ @Test
+ public void testOptionsPreservedOnRepeatedEnlistment() throws Exception {
+ MockEndpoint compensate = getMockEndpoint("mock:compensate");
+ compensate.expectedMessageCount(2);
+
+ try {
+ template.sendBody("direct:workflow-repeated", "go");
+ fail("Should throw an exception");
+ } catch (Exception ex) {
+ // OK
+ }
+
+ compensate.assertIsSatisfied();
+
+ List<String> itemIds = compensate.getReceivedExchanges().stream()
+ .map(e -> e.getMessage().getHeader("itemId", String.class))
+ .sorted()
+ .collect(Collectors.toList());
+ assertEquals(List.of("item-A", "item-B"), itemIds);
+ }
+
@Override
protected RouteBuilder createRouteBuilder() {
@@ -69,6 +95,20 @@ public class SagaOptionsTest extends ContextTestSupport {
context.addService(new InMemorySagaService());
+ from("direct:workflow-repeated").saga()
+ .setHeader("itemId", constant("item-A"))
+ .to("direct:enlistItem")
+ .setHeader("itemId", constant("item-B"))
+ .to("direct:enlistItem")
+ .process(ex -> {
+ throw new RuntimeException("forced failure");
+ });
+
+
from("direct:enlistItem").saga().propagation(SagaPropagation.MANDATORY)
+ .option("itemId", header("itemId"))
+ .compensation("mock:compensate")
+ .log("Enlisted ${header.itemId}");
+
from("direct:workflow").saga().option("id",
constant("myheader")).option("name", header("myname"))
.completion("mock:complete").compensation("mock:compensate")
.choice().when(body().isEqualTo("compensate")).process(ex -> {
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 f6996eda5cfc..85f900ca6dc2 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
@@ -23,7 +23,6 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
@@ -57,16 +56,14 @@ public class InMemorySagaCoordinator implements
CamelSagaCoordinator {
private final CamelContext camelContext;
private final InMemorySagaService sagaService;
private final String sagaId;
- private final List<CamelSagaStep> steps;
- private final Map<CamelSagaStep, Map<String, Object>> optionValues;
+ private final List<StepEnlistment> enlistments;
private final AtomicReference<Status> currentStatus;
public InMemorySagaCoordinator(CamelContext camelContext,
InMemorySagaService sagaService, String sagaId) {
this.camelContext = ObjectHelper.notNull(camelContext, "camelContext");
this.sagaService = ObjectHelper.notNull(sagaService, "sagaService");
this.sagaId = ObjectHelper.notNull(sagaId, "sagaId");
- this.steps = new CopyOnWriteArrayList<>();
- this.optionValues = new ConcurrentHashMap<>();
+ this.enlistments = new CopyOnWriteArrayList<>();
this.currentStatus = new AtomicReference<>(Status.RUNNING);
}
@@ -84,27 +81,26 @@ public class InMemorySagaCoordinator implements
CamelSagaCoordinator {
return res;
}
- this.steps.add(step);
-
+ Map<String, Object> values = new HashMap<>();
if (!step.getOptions().isEmpty()) {
- optionValues.putIfAbsent(step, new ConcurrentHashMap<>());
- Map<String, Object> values = optionValues.computeIfAbsent(step, k
-> new HashMap<>());
- for (String option : step.getOptions().keySet()) {
- Expression expression = step.getOptions().get(option);
+ for (Map.Entry<String, Expression> entry :
step.getOptions().entrySet()) {
+ Expression expression = entry.getValue();
if (expression != null) {
try {
Object value = expression.evaluate(exchange,
Object.class);
if (value != null) {
- values.put(option, value);
+ values.put(entry.getKey(), value);
}
} catch (Exception ex) {
return CompletableFuture.supplyAsync(() -> {
- throw new RuntimeCamelException("Cannot evaluate
saga option '" + option + "'", ex);
+ throw new RuntimeCamelException(
+ "Cannot evaluate saga option '" +
entry.getKey() + "'", ex);
});
}
}
}
}
+ this.enlistments.add(new StepEnlistment(step, values));
if (step.getTimeoutInMilliseconds().isPresent()) {
sagaService.getExecutorService().schedule(() -> {
@@ -188,11 +184,11 @@ public class InMemorySagaCoordinator implements
CamelSagaCoordinator {
final Exchange exchange,
Function<CamelSagaStep, Optional<Endpoint>> endpointExtractor,
String description) {
CompletableFuture<Boolean> result =
CompletableFuture.completedFuture(true);
- for (CamelSagaStep step : reversed(steps)) {
- Optional<Endpoint> endpoint = endpointExtractor.apply(step);
+ for (StepEnlistment enlistment : reversed(enlistments)) {
+ Optional<Endpoint> endpoint =
endpointExtractor.apply(enlistment.step);
if (endpoint.isPresent()) {
result = result.thenCompose(
- prevResult -> doFinalize(exchange, endpoint.get(),
step, 0, description)
+ prevResult -> doFinalize(exchange, endpoint.get(),
enlistment, 0, description)
.thenApply(res -> prevResult && res));
}
}
@@ -206,8 +202,8 @@ public class InMemorySagaCoordinator implements
CamelSagaCoordinator {
}
private CompletableFuture<Boolean> doFinalize(
- Exchange exchange, Endpoint endpoint, CamelSagaStep step, int
doneAttempts, String description) {
- Exchange target = createExchange(exchange, endpoint, step);
+ Exchange exchange, Endpoint endpoint, StepEnlistment enlistment,
int doneAttempts, String description) {
+ Exchange target = createExchange(exchange, endpoint, enlistment);
return CompletableFuture.supplyAsync(() -> {
Exchange res = sagaService.getProducerTemplate().send(endpoint,
target);
@@ -229,7 +225,7 @@ public class InMemorySagaCoordinator implements
CamelSagaCoordinator {
} else {
CompletableFuture<Boolean> future = new CompletableFuture<>();
sagaService.getExecutorService().schedule(() -> {
- doFinalize(target, endpoint, step, currentAttempt,
description).whenComplete((res, ex) -> {
+ doFinalize(target, endpoint, enlistment, currentAttempt,
description).whenComplete((res, ex) -> {
if (ex != null) {
future.completeExceptionally(ex);
} else {
@@ -243,7 +239,7 @@ public class InMemorySagaCoordinator implements
CamelSagaCoordinator {
}
@SuppressWarnings("deprecation")
- private Exchange createExchange(Exchange parent, Endpoint endpoint,
CamelSagaStep step) {
+ private Exchange createExchange(Exchange parent, Endpoint endpoint,
StepEnlistment enlistment) {
Exchange answer = endpoint.createExchange();
answer.getMessage().setHeader(Exchange.SAGA_LONG_RUNNING_ACTION,
getId());
answer.getExchangeExtension().setSagaLongRunningAction(getId());
@@ -254,11 +250,8 @@ public class InMemorySagaCoordinator implements
CamelSagaCoordinator {
answer.setProperty(ExchangePropertyKey.OTEL_ACTIVE_SPAN, span);
}
- Map<String, Object> values = optionValues.get(step);
- if (values != null) {
- for (Map.Entry<String, Object> entry : values.entrySet()) {
- answer.getMessage().setHeader(entry.getKey(),
entry.getValue());
- }
+ for (Map.Entry<String, Object> entry :
enlistment.optionValues.entrySet()) {
+ answer.getMessage().setHeader(entry.getKey(), entry.getValue());
}
return answer;
}
@@ -268,4 +261,14 @@ public class InMemorySagaCoordinator implements
CamelSagaCoordinator {
Collections.reverse(reversed);
return reversed;
}
+
+ private static class StepEnlistment {
+ final CamelSagaStep step;
+ final Map<String, Object> optionValues;
+
+ StepEnlistment(CamelSagaStep step, Map<String, Object> optionValues) {
+ this.step = step;
+ this.optionValues = optionValues;
+ }
+ }
}
