This is an automated email from the ASF dual-hosted git repository. Croway pushed a commit to branch camel-4.18.x in repository https://gitbox.apache.org/repos/asf/camel.git
commit af1efbcccf5882b1ef0094db718972da3190e7e9 Author: Andrea Cosentino <[email protected]> AuthorDate: Mon Aug 31 10:24:08 2026 +0200 CAMEL-24449: camel-core - only consult Long-Running-Action for a saga service that uses it SagaProcessor.getCurrentSagaCoordinator() fell back to the unprefixed Long-Running-Action message header whenever the exchange's internal saga state was missing, so it could pick up a coordinator id from an unrelated caller and join that exchange to the wrong saga under AUTO completion. That fallback was added for LRA protocol interoperability (CAMEL-23469), but applied unconditionally to every saga service, including the default InMemorySagaService, which has no external coordinator to interoperate with. Adds CamelSagaService.isLongRunningActionHeaderSupported(), defaulting to false, and only consults the header when the configured service opts in. LRASagaService overrides it to true, preserving the existing interoperability. A custom CamelSagaService joining sagas via the header must now override this method. KafkaSagaIT is updated to advertise support since its saga id only survives a Kafka round-trip through the header. Includes an upgrade-guide entry for 4.23. Closes #25828 (cherry picked from commit dfde003151a0753f997f9eea7d071097f4b00db2) Co-authored-by: Claude Opus 5 (1M context) <[email protected]> --- .../component/kafka/integration/KafkaSagaIT.java | 16 +++- .../apache/camel/service/lra/LRASagaService.java | 9 ++ .../apache/camel/processor/saga/SagaProcessor.java | 7 +- .../SagaHeaderCannotSelectCoordinatorTest.java | 99 ++++++++++++++++++++++ .../org/apache/camel/saga/CamelSagaService.java | 18 ++++ 5 files changed, 146 insertions(+), 3 deletions(-) diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaSagaIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaSagaIT.java index 187477f4424b..1c0073220d7d 100644 --- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaSagaIT.java +++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaSagaIT.java @@ -45,7 +45,7 @@ public class KafkaSagaIT extends BaseKafkaTestSupport { return new RouteBuilder() { @Override public void configure() throws Exception { - getCamelContext().addService(new InMemorySagaService()); + getCamelContext().addService(new KafkaInteropSagaService()); from("direct:saga") .saga() @@ -64,6 +64,20 @@ public class KafkaSagaIT extends BaseKafkaTestSupport { } } +/** + * The saga id is stored in the exchange's internal state, which does not survive the Kafka produce/consume round-trip - + * only the {@code Long-Running-Action} header does. Advertise header support so the consumer route can join the saga + * started by the producer route, as documented for a {@code CamelSagaService} that relies on the header to join sagas + * started by another participant. + */ +final class KafkaInteropSagaService extends InMemorySagaService { + + @Override + public boolean isLongRunningActionHeaderSupported() { + return true; + } +} + final class SagaBean { public static String id; public static Boolean isSame = false; diff --git a/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaService.java b/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaService.java index e10eed2b07c9..0d3acb834eb5 100644 --- a/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaService.java +++ b/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaService.java @@ -66,6 +66,15 @@ public class LRASagaService extends ServiceSupport implements StaticService, Cam .thenApply(url -> new LRASagaCoordinator(LRASagaService.this, url)); } + /** + * The LRA protocol carries the coordinator id in the {@code Long-Running-Action} header, so a saga started by + * another participant is joined through it. + */ + @Override + public boolean isLongRunningActionHeaderSupported() { + return true; + } + @Override public CompletableFuture<CamelSagaCoordinator> getSaga(String id) { CompletableFuture<CamelSagaCoordinator> coordinator; diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/SagaProcessor.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/SagaProcessor.java index 1e28a1f3a4fb..02a0aefb65bf 100644 --- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/SagaProcessor.java +++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/SagaProcessor.java @@ -54,8 +54,11 @@ public abstract class SagaProcessor extends BaseDelegateProcessorSupport impleme protected CompletableFuture<CamelSagaCoordinator> getCurrentSagaCoordinator(Exchange exchange) { // try internal state first (survives removeHeaders("*")) String currentSaga = exchange.getExchangeExtension().getSagaLongRunningAction(); - if (currentSaga == null) { - // fall back to header for interoperability (e.g., LRA protocol) + if (currentSaga == null && sagaService.isLongRunningActionHeaderSupported()) { + // fall back to header for interoperability (e.g., LRA protocol), but only for a service that takes part + // in such a protocol. Long-Running-Action is outside the Camel namespace that consumers filter, and the + // id is written back onto responses, so consulting it where no external coordinator exists would let a + // message pick which saga its exchange joins. currentSaga = exchange.getIn().getHeader(Exchange.SAGA_LONG_RUNNING_ACTION, String.class); } if (currentSaga != null) { diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/SagaHeaderCannotSelectCoordinatorTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/SagaHeaderCannotSelectCoordinatorTest.java new file mode 100644 index 000000000000..a9487b289e86 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/SagaHeaderCannotSelectCoordinatorTest.java @@ -0,0 +1,99 @@ +/* + * 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.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.model.SagaPropagation; +import org.apache.camel.saga.CamelSagaCoordinator; +import org.apache.camel.saga.CamelSagaStep; +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.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * The saga id normally travels in the exchange's internal state. It is also readable from the + * {@code Long-Running-Action} header so a coordinator started elsewhere can be joined, which is how the LRA protocol + * carries it - but that header sits outside the {@code Camel} namespace consumers filter, and the id is written back + * onto responses, so under a service with no external coordinator it would let a message choose which saga its exchange + * joins. + * <p> + * Asserted by watching which ids reach {@code getSaga}, rather than by joining a live saga: a saga started by another + * route has already completed by the time a second exchange could present its id, so that would fail for the wrong + * reason. + */ +public class SagaHeaderCannotSelectCoordinatorTest extends ContextTestSupport { + + private final RecordingSagaService sagaService = new RecordingSagaService(); + + @Test + public void aMessageSuppliedIdIsNotLookedUp() { + Exchange exchange = template.request("direct:mandatory", e -> { + e.getIn().setHeader(Exchange.SAGA_LONG_RUNNING_ACTION, "a-saga-id-from-the-wire"); + e.getIn().setBody("hello"); + }); + + assertTrue(exchange.isFailed(), "MANDATORY has no saga to join, so the exchange must fail"); + assertEquals(List.of(), sagaService.lookedUp, + "the coordinator must not be looked up from an id supplied by the message"); + } + + @Test + public void theInMemoryServiceDoesNotAdvertiseHeaderSupport() { + assertFalse(new InMemorySagaService().isLongRunningActionHeaderSupported()); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() throws Exception { + context.addService(sagaService); + + from("direct:mandatory").saga().propagation(SagaPropagation.MANDATORY) + .transform().constant("joined"); + } + }; + } + + /** + * Delegates to the in-memory service, recording every id it is asked to resolve. + */ + private static final class RecordingSagaService extends InMemorySagaService { + + private final List<String> lookedUp = new CopyOnWriteArrayList<>(); + + @Override + public CompletableFuture<CamelSagaCoordinator> getSaga(String id) { + lookedUp.add(id); + return super.getSaga(id); + } + + @Override + public void registerStep(CamelSagaStep step) { + super.registerStep(step); + } + } +} diff --git a/core/camel-support/src/main/java/org/apache/camel/saga/CamelSagaService.java b/core/camel-support/src/main/java/org/apache/camel/saga/CamelSagaService.java index e37201da7c56..c7e97bb0ee84 100644 --- a/core/camel-support/src/main/java/org/apache/camel/saga/CamelSagaService.java +++ b/core/camel-support/src/main/java/org/apache/camel/saga/CamelSagaService.java @@ -33,4 +33,22 @@ public interface CamelSagaService extends Service, CamelContextAware { void registerStep(CamelSagaStep step); + /** + * Whether a saga coordinator may be selected from the {@code Long-Running-Action} message header. + * <p> + * The saga id normally travels in the exchange's internal state, which survives {@code removeHeaders("*")}. The + * header is consulted as well so that a coordinator started elsewhere can be joined - the LRA protocol carries the + * id that way. That only makes sense for a service which actually participates in such a protocol: the header sits + * outside the {@code Camel} namespace that consumers filter, and the id is written back onto responses, so where no + * external coordinator exists it lets a message choose which saga its exchange joins. + * <p> + * Defaults to false. A service that takes part in a distributed saga protocol overrides it. + * + * @return true if the {@code Long-Running-Action} header may select a coordinator + * @since 4.23 + */ + default boolean isLongRunningActionHeaderSupported() { + return false; + } + }
