This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24148 in repository https://gitbox.apache.org/repos/asf/camel.git
commit a1ce0273f7f7c14c8a16351f66b950f729adbcbb Author: Claus Ibsen <[email protected]> AuthorDate: Fri Jul 17 10:46:21 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]> --- .../apache/camel/processor/SagaOptionsTest.java | 40 ++++++++++++++++ .../apache/camel/saga/InMemorySagaCoordinator.java | 55 ++++++++++++---------- .../org/apache/camel/saga/InMemorySagaService.java | 14 ++++++ 3 files changed, 83 insertions(+), 26 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 4811325c9d88..212295bf8781 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(() -> { @@ -174,11 +170,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)); } } @@ -192,11 +188,11 @@ 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 = camelContext.createFluentProducerTemplate().to(endpoint).withExchange(target).send(); + Exchange res = sagaService.getProducerTemplate().send(endpoint, target); Exception ex = res.getException(); if (ex != null) { throw new RuntimeCamelException(res.getException()); @@ -215,7 +211,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 { @@ -229,7 +225,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()); @@ -240,11 +236,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; } @@ -254,4 +247,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; + } + } } diff --git a/core/camel-support/src/main/java/org/apache/camel/saga/InMemorySagaService.java b/core/camel-support/src/main/java/org/apache/camel/saga/InMemorySagaService.java index 6b826df3716b..79ceb583f18b 100644 --- a/core/camel-support/src/main/java/org/apache/camel/saga/InMemorySagaService.java +++ b/core/camel-support/src/main/java/org/apache/camel/saga/InMemorySagaService.java @@ -23,6 +23,7 @@ import java.util.concurrent.ScheduledExecutorService; import org.apache.camel.CamelContext; import org.apache.camel.Exchange; +import org.apache.camel.ProducerTemplate; import org.apache.camel.support.service.ServiceSupport; import org.apache.camel.util.ObjectHelper; @@ -41,6 +42,8 @@ public class InMemorySagaService extends ServiceSupport implements CamelSagaServ private ScheduledExecutorService executorService; + private ProducerTemplate producerTemplate; + private int maxRetryAttempts = DEFAULT_MAX_RETRY_ATTEMPTS; private long retryDelayInMilliseconds = DEFAULT_RETRY_DELAY_IN_MILLISECONDS; @@ -72,10 +75,17 @@ public class InMemorySagaService extends ServiceSupport implements CamelSagaServ this.executorService = camelContext.getExecutorServiceManager() .newDefaultScheduledThreadPool(this, "saga"); } + if (this.producerTemplate == null) { + this.producerTemplate = camelContext.createProducerTemplate(); + } } @Override protected void doStop() throws Exception { + if (this.producerTemplate != null) { + this.producerTemplate.stop(); + this.producerTemplate = null; + } if (this.executorService != null) { camelContext.getExecutorServiceManager().shutdownGraceful(this.executorService); this.executorService = null; @@ -86,6 +96,10 @@ public class InMemorySagaService extends ServiceSupport implements CamelSagaServ return executorService; } + public ProducerTemplate getProducerTemplate() { + return producerTemplate; + } + @Override public void setCamelContext(CamelContext camelContext) { this.camelContext = camelContext;
