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 1bd24cd41e69 CAMEL-25419: Saga EIP - keep the original exception when
compensation fails (#27510)
1bd24cd41e69 is described below
commit 1bd24cd41e69d20352b869563f6c7879e6ba8366
Author: Federico Mariani <[email protected]>
AuthorDate: Thu Oct 8 10:46:04 2026 +0200
CAMEL-25419: Saga EIP - keep the original exception when compensation fails
(#27510)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../org/apache/camel/catalog/docs/saga-eip.adoc | 4 +-
.../src/main/docs/modules/eips/pages/saga-eip.adoc | 4 +-
.../apache/camel/processor/saga/SagaProcessor.java | 36 ++++-
.../processor/SagaCompensationFailureTest.java | 166 +++++++++++++++++++++
.../apache/camel/saga/InMemorySagaCoordinator.java | 14 +-
.../org/apache/camel/saga/InMemorySagaService.java | 10 ++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 19 +++
7 files changed, 238 insertions(+), 15 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/saga-eip.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/saga-eip.adoc
index 0b299d89c359..2f8fe11fd646 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/saga-eip.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/saga-eip.adoc
@@ -357,7 +357,9 @@ periodically retry to execute the compensating action up to
a certain limit.
This means that *any compensating action must be idempotent*,
so it should take into account that it may be triggered multiple times and
should not fail in any case.
-If compensation cannot be done after all retries, a manual intervention
process should be triggered by the Saga implementation.
+If compensation cannot be done after all retries, a manual intervention
process should be triggered.
+The in-memory Saga service logs a warning, marks the Saga as failed, and keeps
the exception that caused the compensation
+on the exchange, with the compensation failure added to it as a suppressed
exception.
[NOTE]
====
diff --git
a/core/camel-core-engine/src/main/docs/modules/eips/pages/saga-eip.adoc
b/core/camel-core-engine/src/main/docs/modules/eips/pages/saga-eip.adoc
index 0b299d89c359..2f8fe11fd646 100644
--- a/core/camel-core-engine/src/main/docs/modules/eips/pages/saga-eip.adoc
+++ b/core/camel-core-engine/src/main/docs/modules/eips/pages/saga-eip.adoc
@@ -357,7 +357,9 @@ periodically retry to execute the compensating action up to
a certain limit.
This means that *any compensating action must be idempotent*,
so it should take into account that it may be triggered multiple times and
should not fail in any case.
-If compensation cannot be done after all retries, a manual intervention
process should be triggered by the Saga implementation.
+If compensation cannot be done after all retries, a manual intervention
process should be triggered.
+The in-memory Saga service logs a warning, marks the Saga as failed, and keeps
the exception that caused the compensation
+on the exchange, with the compensation failure added to it as a suppressed
exception.
[NOTE]
====
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 efcb288144e9..dbd650dcb6df 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
@@ -17,6 +17,7 @@
package org.apache.camel.processor.saga;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
@@ -110,19 +111,30 @@ public abstract class SagaProcessor extends
BaseDelegateProcessorSupport
if (this.completionMode == SagaCompletionMode.AUTO) {
if (exchange.getException() != null) {
if (coordinator != null) {
- coordinator.compensate(exchange).whenComplete((done, ex)
-> ifNotException(ex, exchange, callback, () -> {
- setCurrentSagaCoordinator(exchange,
previousCoordinator);
- callback.done(false);
- }));
+ final Exception cause = exchange.getException();
+ coordinator.compensate(exchange).whenComplete((done, ex)
-> {
+ if (ex != null) {
+ // keep the exception that caused the
compensation, and attach the compensation failure to it.
+ // This adds to the exception instance thrown by
the route: if a route throws a shared
+ // instance, its suppressed exceptions grow with
every failed compensation
+ cause.addSuppressed(unwrap(ex));
+ exchange.setException(cause);
+ callback.done(false);
+ } else {
+ setCurrentSagaCoordinator(exchange,
previousCoordinator);
+ callback.done(false);
+ }
+ });
} else {
// No coordinator available, so no saga available.
callback.done(false);
}
} else {
- coordinator.complete(exchange).whenComplete((done, ex) ->
ifNotException(ex, exchange, callback, () -> {
- setCurrentSagaCoordinator(exchange, previousCoordinator);
- callback.done(false);
- }));
+ coordinator.complete(exchange)
+ .whenComplete((done, ex) -> ifNotException(ex != null
? unwrap(ex) : null, exchange, callback, () -> {
+ setCurrentSagaCoordinator(exchange,
previousCoordinator);
+ callback.done(false);
+ }));
}
} else if (this.completionMode == SagaCompletionMode.MANUAL) {
// Completion will be handled manually by the user
@@ -132,6 +144,14 @@ public abstract class SagaProcessor extends
BaseDelegateProcessorSupport
}
}
+ /**
+ * The failure of a dependent stage of a {@link CompletableFuture} is
wrapped in a {@link CompletionException}: the
+ * exchange gets the actual failure.
+ */
+ private static Throwable unwrap(Throwable ex) {
+ return ex instanceof CompletionException && ex.getCause() != null ?
ex.getCause() : ex;
+ }
+
public CamelSagaService getSagaService() {
return sagaService;
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/SagaCompensationFailureTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/SagaCompensationFailureTest.java
new file mode 100644
index 000000000000..49c301709787
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/SagaCompensationFailureTest.java
@@ -0,0 +1,166 @@
+/*
+ * 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.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.log.ConsumingAppender;
+import org.apache.camel.model.SagaCompletionMode;
+import org.apache.camel.saga.CamelSagaCoordinator;
+import org.apache.camel.saga.InMemorySagaService;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * CAMEL-25419: what the saga EIP reports when a compensation or a completion
cannot be done, and what the in-memory
+ * saga service logs when it stops with sagas that are not finalized.
+ */
+public class SagaCompensationFailureTest extends ContextTestSupport {
+
+ private static final String SAGA_SERVICE_LOGGER =
InMemorySagaService.class.getName();
+
+ private final AtomicReference<CamelSagaCoordinator> coordinator = new
AtomicReference<>();
+ private final AtomicReference<String> manualSagaId = new
AtomicReference<>();
+ private final List<String> sagaServiceWarnings = new
CopyOnWriteArrayList<>();
+
+ @Test
+ public void testFailedCompensationKeepsTheOriginalException() {
+ Exchange result = template.send("direct:saga-fails", e ->
e.getMessage().setBody("hello"));
+
+ // the exception that caused the compensation stays on the exchange
+ Exception ex = result.getException();
+ assertInstanceOf(IllegalArgumentException.class, ex);
+ assertEquals("business failure", ex.getMessage());
+
+ // and the compensation failure is attached to it
+ Throwable[] suppressed = ex.getSuppressed();
+ assertEquals(1, suppressed.length, "the compensation failure should be
attached as a suppressed exception");
+ assertInstanceOf(RuntimeCamelException.class, suppressed[0]);
+ assertTrue(suppressed[0].getMessage().contains("Unable to compensate
all required steps of the saga"),
+ suppressed[0].getMessage());
+ }
+
+ @Test
+ public void testFailedCompensationIsNotReportedAsCompensated() throws
Exception {
+ Exchange result = template.send("direct:saga-fails", e ->
e.getMessage().setBody("hello"));
+ assertNotNull(result.getException());
+
+ CamelSagaCoordinator saga = coordinator.get();
+ assertNotNull(saga);
+ // the compensation could not be done: compensating the saga again
must not report success
+ ExecutionException ex = assertThrows(ExecutionException.class,
+ () -> saga.compensate(result).get(5, TimeUnit.SECONDS),
+ "a saga whose compensation failed is reported as successfully
compensated");
+ assertInstanceOf(IllegalStateException.class, ex.getCause());
+ assertTrue(ex.getCause().getMessage().contains("status is FAILED"),
ex.getCause().getMessage());
+ }
+
+ @Test
+ public void testFailedCompletionIsNotWrapped() {
+ Exchange result = template.send("direct:saga-completion-fails", e ->
e.getMessage().setBody("hello"));
+
+ // the completion failure is on the exchange as is, not wrapped in a
CompletionException
+ Exception ex = result.getException();
+ assertInstanceOf(RuntimeCamelException.class, ex);
+ assertTrue(ex.getMessage().contains("Unable to complete all required
steps of the saga"), ex.getMessage());
+ }
+
+ @Test
+ public void testSagaNotFinalizedWhenContextStopsIsLogged() throws
Exception {
+ // a saga in MANUAL completion mode that is never completed nor
compensated
+ template.sendBody("direct:saga-manual", "hello");
+ String sagaId = manualSagaId.get();
+ assertNotNull(sagaId);
+
+ // the in-memory service does not persist sagas: stopping abandons the
saga, which must be logged
+ context.stop();
+
+ assertTrue(sagaServiceWarnings.stream().anyMatch(m ->
m.contains(sagaId)),
+ "no warning names the abandoned saga " + sagaId + ": " +
sagaServiceWarnings);
+ }
+
+ @Override
+ @BeforeEach
+ public void setUp() throws Exception {
+ ConsumingAppender.newAppender(SAGA_SERVICE_LOGGER,
"SagaCompensationFailureTest", Level.WARN,
+ e ->
sagaServiceWarnings.add(e.getMessage().getFormattedMessage()));
+ super.setUp();
+ }
+
+ @Override
+ @AfterEach
+ public void tearDown() throws Exception {
+ super.tearDown();
+ LoggerContext ctx = (LoggerContext) LogManager.getContext(false);
+ ctx.getConfiguration().removeLogger(SAGA_SERVICE_LOGGER);
+ ctx.updateLoggers();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ InMemorySagaService sagaService = new InMemorySagaService();
+ sagaService.setMaxRetryAttempts(2);
+ sagaService.setRetryDelayInMilliseconds(100);
+ context.addService(sagaService);
+
+ from("direct:saga-fails")
+ .saga().compensation("direct:finalization-fails")
+ .process(e -> coordinator.set(
+
sagaService.getSaga(e.getExchangeExtension().getSagaLongRunningAction()).get()))
+ .process(e -> {
+ throw new IllegalArgumentException("business
failure");
+ });
+
+ from("direct:saga-completion-fails")
+ .saga().completion("direct:finalization-fails")
+ .to("mock:end");
+
+ from("direct:finalization-fails")
+ .process(e -> {
+ throw new IllegalStateException("finalization
failure");
+ });
+
+ from("direct:saga-manual")
+ .saga().completionMode(SagaCompletionMode.MANUAL)
+ .compensation("mock:compensation")
+ .process(e ->
manualSagaId.set(e.getExchangeExtension().getSagaLongRunningAction()))
+ .to("mock:end");
+ }
+ };
+ }
+}
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 a91fed79609e..b12887f9cd2e 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
@@ -51,7 +51,9 @@ public class InMemorySagaCoordinator implements
CamelSagaCoordinator {
COMPENSATING,
COMPENSATED,
COMPLETING,
- COMPLETED
+ COMPLETED,
+ // the compensation or completion could not be done for every step,
after all the retries
+ FAILED
}
private static final Logger LOG =
LoggerFactory.getLogger(InMemorySagaCoordinator.class);
@@ -209,10 +211,11 @@ public class InMemorySagaCoordinator implements
CamelSagaCoordinator {
private CompletableFuture<Boolean> doCompensate(final Exchange exchange,
List<StepEnlistment> steps) {
return doFinalize(exchange, steps, CamelSagaStep::getCompensation,
"compensation")
.whenComplete((res, ex) -> {
- if (ex != null || !Boolean.TRUE.equals(res)) {
+ boolean failed = ex != null || !Boolean.TRUE.equals(res);
+ if (failed) {
LOG.warn("Saga {} compensation did not fully succeed —
manual intervention may be needed", sagaId);
}
- currentStatus.set(Status.COMPENSATED);
+ currentStatus.set(failed ? Status.FAILED :
Status.COMPENSATED);
sagaService.removeSaga(sagaId);
});
}
@@ -224,10 +227,11 @@ public class InMemorySagaCoordinator implements
CamelSagaCoordinator {
private CompletableFuture<Boolean> doComplete(final Exchange exchange,
List<StepEnlistment> steps) {
return doFinalize(exchange, steps, CamelSagaStep::getCompletion,
"completion")
.whenComplete((res, ex) -> {
- if (ex != null || !Boolean.TRUE.equals(res)) {
+ boolean failed = ex != null || !Boolean.TRUE.equals(res);
+ if (failed) {
LOG.warn("Saga {} completion did not fully succeed —
manual intervention may be needed", sagaId);
}
- currentStatus.set(Status.COMPLETED);
+ currentStatus.set(failed ? Status.FAILED :
Status.COMPLETED);
sagaService.removeSaga(sagaId);
});
}
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 11b55c387a82..bdeef298e2db 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
@@ -26,12 +26,16 @@ import org.apache.camel.Exchange;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.support.service.ServiceSupport;
import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* An in-memory implementation of a saga service.
*/
public class InMemorySagaService extends ServiceSupport implements
CamelSagaService {
+ private static final Logger LOG =
LoggerFactory.getLogger(InMemorySagaService.class);
+
public static final int DEFAULT_MAX_RETRY_ATTEMPTS = 5;
public static final long DEFAULT_RETRY_DELAY_IN_MILLISECONDS = 5000;
@@ -86,6 +90,12 @@ public class InMemorySagaService extends ServiceSupport
implements CamelSagaServ
@Override
protected void doStop() throws Exception {
+ if (!coordinators.isEmpty()) {
+ // the saga status is not persisted: these sagas will not be
compensated or completed
+ LOG.warn("Stopping the in-memory saga service with {} saga(s) not
finalized, which will not be compensated"
+ + " or completed: {}",
+ coordinators.size(), coordinators.keySet());
+ }
if (this.producerTemplate != null) {
this.producerTemplate.stop();
this.producerTemplate = null;
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
index 5f179794fc64..5236350ed66e 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
@@ -326,6 +326,25 @@ When a route with the stream resequencer stops, a caller
that is blocked because
now fails with a `java.util.concurrent.RejectedExecutionException`. Prior to
Camel 4.23 such a caller stayed blocked,
also after the route was stopped.
+=== Saga EIP - compensation and completion failures
+
+When a saga in `AUTO` completion mode fails and its compensation then fails as
well, the exchange now keeps the
+exception that caused the compensation, and the compensation failure (a
`RuntimeCamelException` with the message
+`Unable to compensate all required steps of the saga <id>`) is added to it as
a suppressed exception.
+Prior to Camel 4.23 that `RuntimeCamelException`, wrapped in a
`java.util.concurrent.CompletionException`, replaced
+the original exception. An `onException` or `doCatch` that matched the
`CompletionException`, the
+`RuntimeCamelException` or its message must now match the original exception
instead, and can find the compensation
+failure in its suppressed exceptions.
+
+When the completion of a saga fails, the exchange now gets the
`RuntimeCamelException` with the message
+`Unable to complete all required steps of the saga <id>` as is. Prior to Camel
4.23 it was wrapped in a
+`java.util.concurrent.CompletionException`.
+
+With the in-memory saga service, a saga whose compensation or completion could
not be done after all the retries now
+has the status `FAILED`, instead of `COMPENSATED` or `COMPLETED`: compensating
or completing it again fails, instead of
+reporting success. When the in-memory saga service stops, it logs a warning
with the ids of the sagas that are not
+finalized, since they will not be compensated or completed.
+
=== Tokenize language
- A message without a body (or a `source` that has no value) now has no
tokens, the same as an empty body.
