CAMEL-24150, CAMEL-24151: Saga EIP fixes — shared service lifecycle, code review findings

components/camel-lra/src/main/java/org/apache/camel/service/lra/LRAClient.java

@@ -125,7 +125,8 @@ public CompletableFuture<Void> join(final URL lra, LRASagaStep step, Exchange ex
 
             String lraEndpoint = lra.toString();
             if (step.getTimeoutInMilliseconds().isPresent()) {
-                lraEndpoint = lraEndpoint + "?" + HEADER_TIME_LIMIT + "=" + step.getTimeoutInMilliseconds().get();
+                String separator = lraEndpoint.contains("?") ? "&" : "?";
+                lraEndpoint = lraEndpoint + separator + HEADER_TIME_LIMIT + "=" + step.getTimeoutInMilliseconds().get();
             }
             HttpRequest request = prepareRequest(URI.create(lraEndpoint), exchange)
                     .setHeader(HEADER_LINK, link.toString())
@@ -138,8 +139,10 @@ public CompletableFuture<Void> join(final URL lra, LRASagaStep step, Exchange ex
         }, sagaService.getExecutorService())
                 .thenCompose(Function.identity())
                 .thenApply(response -> {
-                    if (response.statusCode() != HttpURLConnection.HTTP_OK) {
-                        throw new RuntimeCamelException("Cannot join LRA");
+                    int status = response.statusCode();
+                    if (status >= HttpURLConnection.HTTP_BAD_REQUEST) {
+                        throw new RuntimeCamelException(
+                                "Cannot join LRA " + lra + " (HTTP " + status + "): " + response.body());
                     }
 
                     return null;
@@ -155,8 +158,10 @@ public CompletableFuture<Void> complete(URL lra, Exchange exchange) {
         CompletableFuture<HttpResponse<String>> future = client.sendAsync(request, HttpResponse.BodyHandlers.ofString());
 
         return future.thenApply(response -> {
-            if (response.statusCode() != HttpURLConnection.HTTP_OK) {
-                throw new RuntimeCamelException("Cannot complete LRA");
+            int status = response.statusCode();
+            if (status >= HttpURLConnection.HTTP_BAD_REQUEST) {
+                throw new RuntimeCamelException(
+                        "Cannot complete LRA " + lra + " (HTTP " + status + "): " + response.body());
             }
 
             return null;
@@ -172,8 +177,10 @@ public CompletableFuture<Void> compensate(URL lra, Exchange exchange) {
         CompletableFuture<HttpResponse<String>> future = client.sendAsync(request, HttpResponse.BodyHandlers.ofString());
 
         return future.thenApply(response -> {
-            if (response.statusCode() != HttpURLConnection.HTTP_OK) {
-                throw new RuntimeCamelException("Cannot compensate LRA");
+            int status = response.statusCode();
+            if (status >= HttpURLConnection.HTTP_BAD_REQUEST) {
+                throw new RuntimeCamelException(
+                        "Cannot compensate LRA " + lra + " (HTTP " + status + "): " + response.body());
             }
 
             return null;

components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaCoordinator.java

@@ -41,9 +41,7 @@ public CompletableFuture<Void> beginStep(Exchange exchange, CamelSagaStep step)
         try {
             sagaStep = LRASagaStep.fromCamelSagaStep(step, exchange);
         } catch (RuntimeException ex) {
-            return CompletableFuture.supplyAsync(() -> {
-                throw ex;
-            });
+            return CompletableFuture.failedFuture(ex);
         }
         return sagaService.getClient().join(this.lraURL, sagaStep, exchange);
     }

components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaService.java

@@ -79,13 +79,23 @@ public CompletableFuture<CamelSagaCoordinator> getSaga(String id) {
 
     @Override
     public void registerStep(CamelSagaStep step) {
-        // Register which uris should be exposed
-        step.getCompensation().map(Endpoint::getEndpointUri).map(this.sagaURIs::add);
-        step.getCompletion().map(Endpoint::getEndpointUri).map(this.sagaURIs::add);
+        step.getCompensation().map(Endpoint::getEndpointUri).ifPresent(this.sagaURIs::add);
+        step.getCompletion().map(Endpoint::getEndpointUri).ifPresent(this.sagaURIs::add);
     }
 
     @Override
     protected void doStart() throws Exception {
+        if (coordinatorUrl == null) {
+            throw new IllegalStateException("coordinatorUrl must be configured on the LRA saga service");
+        }
+        if (localParticipantUrl == null) {
+            throw new IllegalStateException("localParticipantUrl must be configured on the LRA saga service");
+        }
+
+        if (this.routes == null) {
+            this.routes = new LRASagaRoutes(this);
+            camelContext.addRoutes(this.routes);
+        }
         if (this.executorService == null) {
             this.executorService = camelContext.getExecutorServiceManager()
                     .newDefaultScheduledThreadPool(this, "saga-lra");
@@ -120,14 +130,6 @@ protected void doStop() throws Exception {
     @Override
     public void setCamelContext(CamelContext camelContext) {
         this.camelContext = camelContext;
-        if (this.routes == null) {
-            this.routes = new LRASagaRoutes(this);
-            try {
-                this.camelContext.addRoutes(this.routes);
-            } catch (Exception ex) {
-                throw RuntimeCamelException.wrapRuntimeException(ex);
-            }
-        }
     }
 
     @Override

core/camel-core-engine/src/main/docs/modules/eips/pages/saga-eip.adoc

@@ -711,7 +711,9 @@ Sagas are long-running actions, but this does not mean that they should not have
 
 [NOTE]
 ====
-The Saga EIP implementation may have a default timeout set on all Sagas that don't specify it explicitly
+There is no default timeout on Sagas.
+If no timeout is specified, a Saga may remain open indefinitely in the case of failure.
+Always set an explicit timeout, especially when using `MANUAL` completion mode.
 ====
 
 When the timeout expires, the Saga EIP will decide to *cancel the Saga* (and compensate all participants), unless a different decision has been taken before.

core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/MandatorySagaProcessor.java

@@ -17,7 +17,6 @@
 package org.apache.camel.processor.saga;
 
 import org.apache.camel.AsyncCallback;
-import org.apache.camel.CamelContext;
 import org.apache.camel.CamelExchangeException;
 import org.apache.camel.Exchange;
 import org.apache.camel.Processor;
@@ -29,9 +28,9 @@
  */
 public class MandatorySagaProcessor extends SagaProcessor {
 
-    public MandatorySagaProcessor(CamelContext camelContext, Processor childProcessor, CamelSagaService sagaService,
+    public MandatorySagaProcessor(Processor childProcessor, CamelSagaService sagaService,
                                   SagaCompletionMode completionMode, CamelSagaStep step) {
-        super(camelContext, childProcessor, sagaService, completionMode, step);
+        super(childProcessor, sagaService, completionMode, step);
     }
 
     @Override

core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/NeverSagaProcessor.java

@@ -17,7 +17,6 @@
 package org.apache.camel.processor.saga;
 
 import org.apache.camel.AsyncCallback;
-import org.apache.camel.CamelContext;
 import org.apache.camel.CamelExchangeException;
 import org.apache.camel.Exchange;
 import org.apache.camel.Processor;
@@ -29,9 +28,9 @@
  */
 public class NeverSagaProcessor extends SagaProcessor {
 
-    public NeverSagaProcessor(CamelContext camelContext, Processor childProcessor, CamelSagaService sagaService,
+    public NeverSagaProcessor(Processor childProcessor, CamelSagaService sagaService,
                               SagaCompletionMode completionMode, CamelSagaStep step) {
-        super(camelContext, childProcessor, sagaService, completionMode, step);
+        super(childProcessor, sagaService, completionMode, step);
         if (!step.isEmpty()) {
             throw new IllegalArgumentException("Saga configuration is not allowed when propagation is set to NEVER");
         }

core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/NotSupportedSagaProcessor.java

@@ -17,7 +17,6 @@
 package org.apache.camel.processor.saga;
 
 import org.apache.camel.AsyncCallback;
-import org.apache.camel.CamelContext;
 import org.apache.camel.Exchange;
 import org.apache.camel.Processor;
 import org.apache.camel.saga.CamelSagaService;
@@ -28,9 +27,9 @@
  */
 public class NotSupportedSagaProcessor extends SagaProcessor {
 
-    public NotSupportedSagaProcessor(CamelContext camelContext, Processor childProcessor, CamelSagaService sagaService,
+    public NotSupportedSagaProcessor(Processor childProcessor, CamelSagaService sagaService,
                                      SagaCompletionMode completionMode, CamelSagaStep step) {
-        super(camelContext, childProcessor, sagaService, completionMode, step);
+        super(childProcessor, sagaService, completionMode, step);
         if (!step.isEmpty()) {
             throw new IllegalArgumentException("Saga configuration is not allowed when propagation is set to NOT_SUPPORTED");
         }

core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/RequiredSagaProcessor.java

@@ -19,7 +19,6 @@
 import java.util.concurrent.CompletableFuture;
 
 import org.apache.camel.AsyncCallback;
-import org.apache.camel.CamelContext;
 import org.apache.camel.Exchange;
 import org.apache.camel.Processor;
 import org.apache.camel.saga.CamelSagaCoordinator;
@@ -31,9 +30,9 @@
  */
 public class RequiredSagaProcessor extends SagaProcessor {
 
-    public RequiredSagaProcessor(CamelContext camelContext, Processor childProcessor, CamelSagaService sagaService,
+    public RequiredSagaProcessor(Processor childProcessor, CamelSagaService sagaService,
                                  SagaCompletionMode completionMode, CamelSagaStep step) {
-        super(camelContext, childProcessor, sagaService, completionMode, step);
+        super(childProcessor, sagaService, completionMode, step);
     }
 
     @Override

core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/RequiresNewSagaProcessor.java

@@ -17,7 +17,6 @@
 package org.apache.camel.processor.saga;
 
 import org.apache.camel.AsyncCallback;
-import org.apache.camel.CamelContext;
 import org.apache.camel.Exchange;
 import org.apache.camel.Processor;
 import org.apache.camel.saga.CamelSagaService;
@@ -28,9 +27,9 @@
  */
 public class RequiresNewSagaProcessor extends SagaProcessor {
 
-    public RequiresNewSagaProcessor(CamelContext camelContext, Processor childProcessor, CamelSagaService sagaService,
+    public RequiresNewSagaProcessor(Processor childProcessor, CamelSagaService sagaService,
                                     SagaCompletionMode completionMode, CamelSagaStep step) {
-        super(camelContext, childProcessor, sagaService, completionMode, step);
+        super(childProcessor, sagaService, completionMode, step);
     }
 
     @Override

core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/SagaProcessor.java

@@ -19,7 +19,6 @@
 import java.util.concurrent.CompletableFuture;
 
 import org.apache.camel.AsyncCallback;
-import org.apache.camel.CamelContext;
 import org.apache.camel.Exchange;
 import org.apache.camel.Processor;
 import org.apache.camel.Traceable;
@@ -30,7 +29,6 @@
 import org.apache.camel.spi.IdAware;
 import org.apache.camel.spi.RouteIdAware;
 import org.apache.camel.spi.StepIdAware;
-import org.apache.camel.support.service.ServiceHelper;
 import org.apache.camel.util.ObjectHelper;
 
 /**
@@ -46,7 +44,7 @@ public abstract class SagaProcessor extends BaseDelegateProcessorSupport
     private String routeId;
     private String stepId;
 
-    protected SagaProcessor(CamelContext camelContext, Processor childProcessor, CamelSagaService sagaService,
+    protected SagaProcessor(Processor childProcessor, CamelSagaService sagaService,
                             SagaCompletionMode completionMode, CamelSagaStep step) {
         super(ObjectHelper.notNull(childProcessor, "childProcessor"));
         this.sagaService = ObjectHelper.notNull(sagaService, "sagaService");
@@ -144,7 +142,7 @@ public void setStepId(String stepId) {
 
     @Override
     public String toString() {
-        return "id";
+        return id;
     }
 
     @Override
@@ -167,19 +165,13 @@ protected void ifNotException(
                 callback.done(false);
             }
         } else {
-            code.run();
+            try {
+                code.run();
+            } catch (Exception e) {
+                exchange.setException(e);
+                callback.done(false);
+            }
         }
     }
 
-    @Override
-    protected void doStart() throws Exception {
-        super.doStart();
-        ServiceHelper.startService(sagaService);
-    }
-
-    @Override
-    protected void doStop() throws Exception {
-        super.doStop();
-        ServiceHelper.stopService(sagaService);
-    }
 }

core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/SagaProcessorBuilder.java

@@ -16,7 +16,6 @@
  */
 package org.apache.camel.processor.saga;
 
-import org.apache.camel.CamelContext;
 import org.apache.camel.Processor;
 import org.apache.camel.saga.CamelSagaService;
 import org.apache.camel.saga.CamelSagaStep;
@@ -26,8 +25,6 @@
  */
 public class SagaProcessorBuilder {
 
-    private CamelContext camelContext;
-
     private Processor childProcessor;
 
     private CamelSagaService sagaService;
@@ -41,11 +38,6 @@ public class SagaProcessorBuilder {
     public SagaProcessorBuilder() {
     }
 
-    public SagaProcessorBuilder camelContext(CamelContext camelContext) {
-        this.camelContext = camelContext;
-        return this;
-    }
-
     public SagaProcessorBuilder childProcessor(Processor childProcessor) {
         this.childProcessor = childProcessor;
         return this;
@@ -78,17 +70,17 @@ public SagaProcessor build() {
 
         switch (propagation) {
             case REQUIRED:
-                return new RequiredSagaProcessor(camelContext, childProcessor, sagaService, completionMode, step);
+                return new RequiredSagaProcessor(childProcessor, sagaService, completionMode, step);
             case REQUIRES_NEW:
-                return new RequiresNewSagaProcessor(camelContext, childProcessor, sagaService, completionMode, step);
+                return new RequiresNewSagaProcessor(childProcessor, sagaService, completionMode, step);
             case SUPPORTS:
-                return new SupportsSagaProcessor(camelContext, childProcessor, sagaService, completionMode, step);
+                return new SupportsSagaProcessor(childProcessor, sagaService, completionMode, step);
             case NOT_SUPPORTED:
-                return new NotSupportedSagaProcessor(camelContext, childProcessor, sagaService, completionMode, step);
+                return new NotSupportedSagaProcessor(childProcessor, sagaService, completionMode, step);
             case NEVER:
-                return new NeverSagaProcessor(camelContext, childProcessor, sagaService, completionMode, step);
+                return new NeverSagaProcessor(childProcessor, sagaService, completionMode, step);
             case MANDATORY:
-                return new MandatorySagaProcessor(camelContext, childProcessor, sagaService, completionMode, step);
+                return new MandatorySagaProcessor(childProcessor, sagaService, completionMode, step);
             default:
                 throw new IllegalStateException("Unsupported propagation mode: " + propagation);
         }

core/camel-core-processor/src/main/java/org/apache/camel/processor/saga/SupportsSagaProcessor.java

@@ -17,7 +17,6 @@
 package org.apache.camel.processor.saga;
 
 import org.apache.camel.AsyncCallback;
-import org.apache.camel.CamelContext;
 import org.apache.camel.Exchange;
 import org.apache.camel.Processor;
 import org.apache.camel.saga.CamelSagaService;
@@ -28,9 +27,9 @@
  */
 public class SupportsSagaProcessor extends SagaProcessor {
 
-    public SupportsSagaProcessor(CamelContext camelContext, Processor childProcessor, CamelSagaService sagaService,
+    public SupportsSagaProcessor(Processor childProcessor, CamelSagaService sagaService,
                                  SagaCompletionMode completionMode, CamelSagaStep step) {
-        super(camelContext, childProcessor, sagaService, completionMode, step);
+        super(childProcessor, sagaService, completionMode, step);
         if (completionMode != null && completionMode != SagaCompletionMode.defaultCompletionMode()) {
             throw new IllegalArgumentException("CompletionMode cannot be specified when propagation is SUPPORTS");
         }

core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SagaReifier.java

@@ -36,9 +36,13 @@
 import org.apache.camel.saga.CamelSagaService;
 import org.apache.camel.saga.CamelSagaStep;
 import org.apache.camel.support.EndpointHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class SagaReifier extends ProcessorReifier<SagaDefinition> {
 
+    private static final Logger LOG = LoggerFactory.getLogger(SagaReifier.class);
+
     public SagaReifier(Route route, ProcessorDefinition<?> definition) {
         super(route, (SagaDefinition) definition);
     }
@@ -104,13 +108,19 @@ public Processor createProcessor() throws Exception {
             completionMode = SagaCompletionMode.defaultCompletionMode();
         }
 
+        if (completionMode == SagaCompletionMode.MANUAL && timeout == null) {
+            LOG.warn("Saga in route '{}' uses MANUAL completion without a timeout."
+                     + " The saga will remain open indefinitely if never completed or compensated manually.",
+                    route.getRouteId());
+        }
+
         Processor childProcessor = this.createChildProcessor(true);
         CamelSagaService camelSagaService = resolveSagaService();
         CamelContextAware.trySetCamelContext(camelSagaService, getCamelContext());
 
         camelSagaService.registerStep(step);
 
-        SagaProcessor answer = new SagaProcessorBuilder().camelContext(camelContext).childProcessor(childProcessor)
+        SagaProcessor answer = new SagaProcessorBuilder().childProcessor(childProcessor)
                 .sagaService(camelSagaService).step(step)
                 .propagation(propagation).completionMode(completionMode).build();
         answer.setDisabled(isDisabled(camelContext, definition));