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 a38ed8bea182 CAMEL-24144: InMemorySagaService - Remove coordinators
from map after completion or compensation
a38ed8bea182 is described below
commit a38ed8bea18240267a45a1e78d9b764dae357bd6
Author: Claus Ibsen <[email protected]>
AuthorDate: Fri Jul 17 14:07:57 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]>
Co-authored-by: Claude Opus 4.6 <[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
