This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24144 in repository https://gitbox.apache.org/repos/asf/camel.git
commit d3657e60ca6c55856b2b950097d3e9e76d14dc45 Author: Claus Ibsen <[email protected]> AuthorDate: Fri Jul 17 09:53:01 2026 +0200 CAMEL-24144: InMemorySagaService - Remove coordinators from map after completion or compensation Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../processor/SagaCoordinatorCleanupTest.java | 101 +++++++++++++++++++++ .../apache/camel/processor/SagaTimeoutTest.java | 6 +- .../apache/camel/saga/InMemorySagaCoordinator.java | 2 + .../org/apache/camel/saga/InMemorySagaService.java | 4 + 4 files changed, 110 insertions(+), 3 deletions(-) diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/SagaCoordinatorCleanupTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/SagaCoordinatorCleanupTest.java new file mode 100644 index 000000000000..312ba3922448 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/SagaCoordinatorCleanupTest.java @@ -0,0 +1,101 @@ +/* + * 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.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.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertNull; + +public class SagaCoordinatorCleanupTest extends ContextTestSupport { + + private InMemorySagaService sagaService; + + @Test + public void testCoordinatorRemovedAfterCompletion() throws Exception { + getMockEndpoint("mock:complete").expectedMessageCount(1); + + String sagaId = context.createFluentProducerTemplate() + .to("direct:saga-success") + .request(String.class); + + MockEndpoint.assertIsSatisfied(context); + + await().atMost(5, TimeUnit.SECONDS) + .untilAsserted(() -> assertNull(sagaService.getSaga(sagaId).get(), + "Coordinator should be removed after completion")); + } + + @Test + public void testCoordinatorRemovedAfterCompensation() throws Exception { + getMockEndpoint("mock:compensate").expectedMessageCount(1); + + try { + context.createFluentProducerTemplate() + .to("direct:saga-fail") + .request(); + } catch (Exception e) { + // expected + } + + MockEndpoint.assertIsSatisfied(context); + + String sagaId = getMockEndpoint("mock:compensate").getReceivedExchanges().get(0) + .getMessage().getHeader(Exchange.SAGA_LONG_RUNNING_ACTION, String.class); + + await().atMost(5, TimeUnit.SECONDS) + .untilAsserted(() -> assertNull(sagaService.getSaga(sagaId).get(), + "Coordinator should be removed after compensation")); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() throws Exception { + sagaService = new InMemorySagaService(); + context.addService(sagaService); + + from("direct:saga-success") + .saga().propagation(SagaPropagation.REQUIRES_NEW) + .completion("direct:complete") + .transform().header(Exchange.SAGA_LONG_RUNNING_ACTION); + + from("direct:complete") + .to("mock:complete"); + + from("direct:saga-fail") + .saga().propagation(SagaPropagation.REQUIRES_NEW) + .compensation("direct:compensate") + .process(e -> { + throw new RuntimeException("forced failure"); + }); + + from("direct:compensate") + .to("mock:compensate"); + } + }; + } +} diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/SagaTimeoutTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/SagaTimeoutTest.java index 14b57842abe7..87eebbef1cd0 100644 --- a/core/camel-core/src/test/java/org/apache/camel/processor/SagaTimeoutTest.java +++ b/core/camel-core/src/test/java/org/apache/camel/processor/SagaTimeoutTest.java @@ -84,10 +84,10 @@ public class SagaTimeoutTest extends ContextTestSupport { template.sendBody("direct:saga-multi-participants", "Hello"); }); - String msg1 = "Cannot begin: status is COMPENSATING"; - String msg2 = "Cannot begin: status is COMPENSATED"; String msg = ex.getCause().getMessage(); - assertTrue(msg.equals(msg1) || msg.equals(msg2)); + assertTrue(msg.contains("Cannot begin: status is COMPENSATING") + || msg.contains("Cannot begin: status is COMPENSATED") + || msg.contains("Exchange is not part of a saga")); end.assertIsSatisfied(); complete.assertIsSatisfied(); 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..0a29ad5b6e17 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 @@ -158,6 +158,7 @@ public class InMemorySagaCoordinator implements CamelSagaCoordinator { return doFinalize(exchange, CamelSagaStep::getCompensation, "compensation") .thenApply(res -> { currentStatus.set(Status.COMPENSATED); + sagaService.removeSaga(sagaId); return res; }); } @@ -166,6 +167,7 @@ public class InMemorySagaCoordinator implements CamelSagaCoordinator { return doFinalize(exchange, CamelSagaStep::getCompletion, "completion") .thenApply(res -> { currentStatus.set(Status.COMPLETED); + sagaService.removeSaga(sagaId); return res; }); } 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..1f90d61a21e7 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 @@ -61,6 +61,10 @@ public class InMemorySagaService extends ServiceSupport implements CamelSagaServ return CompletableFuture.completedFuture(coordinators.get(id)); } + void removeSaga(String sagaId) { + coordinators.remove(sagaId); + } + @Override public void registerStep(CamelSagaStep step) { // do nothing
