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 a9949541267e CAMEL-24192: camel-lra - separate compensation/completion
URI allowlists and prune on route removal (#24904)
a9949541267e is described below
commit a9949541267e7f7be5175646e6677ac27ac06f30
Author: Claus Ibsen <[email protected]>
AuthorDate: Sun Jul 19 20:06:17 2026 +0200
CAMEL-24192: camel-lra - separate compensation/completion URI allowlists
and prune on route removal (#24904)
CAMEL-24192: camel-lra - separate compensation/completion URI allowlists
and prune on route removal
Co-Authored-By: Claude Opus 4.6 <[email protected]>
---
.../apache/camel/service/lra/LRASagaRoutes.java | 43 +++---
.../apache/camel/service/lra/LRASagaService.java | 75 ++++++++++-
.../service/lra/LRASagaServiceAllowlistTest.java | 148 +++++++++++++++++++++
.../java/org/apache/camel/reifier/SagaReifier.java | 1 +
.../java/org/apache/camel/saga/CamelSagaStep.java | 10 ++
5 files changed, 254 insertions(+), 23 deletions(-)
diff --git
a/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaRoutes.java
b/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaRoutes.java
index b2538ac31c67..57ae40880e24 100644
---
a/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaRoutes.java
+++
b/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaRoutes.java
@@ -18,7 +18,10 @@ package org.apache.camel.service.lra;
import java.net.URLDecoder;
import java.nio.charset.StandardCharsets;
-import java.util.*;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
import java.util.stream.Collectors;
import org.apache.camel.Exchange;
@@ -44,7 +47,7 @@ public class LRASagaRoutes extends RouteBuilder {
rest(sagaService.getLocalParticipantContextPath())
.put(PARTICIPANT_PATH_COMPENSATE).to("direct:lra-compensation");
from("direct:lra-compensation").routeId("lra-compensation")
- .process(this::verifyRequest)
+ .process(this::verifyCompensationRequest)
.choice()
.when(header(URL_COMPENSATION_KEY).isNotNull())
.toD("${header." + URL_COMPENSATION_KEY + "}")
@@ -53,7 +56,7 @@ public class LRASagaRoutes extends RouteBuilder {
rest(sagaService.getLocalParticipantContextPath())
.put(PARTICIPANT_PATH_COMPLETE).to("direct:lra-completion");
from("direct:lra-completion").routeId("lra-completion")
- .process(this::verifyRequest)
+ .process(this::verifyCompletionRequest)
.choice()
.when(header(URL_COMPLETION_KEY).isNotNull())
.toD("${header." + URL_COMPLETION_KEY + "}")
@@ -93,47 +96,49 @@ public class LRASagaRoutes extends RouteBuilder {
return URLDecoder.decode(encodedString, StandardCharsets.UTF_8);
}
+ private void verifyCompensationRequest(Exchange exchange) {
+ verifyRequest(exchange, true);
+ }
+
+ private void verifyCompletionRequest(Exchange exchange) {
+ verifyRequest(exchange, false);
+ }
+
/**
- * Check if the request is pointing to an allowed URI to prevent
unauthorized remote uri invocation
+ * Check if the request is pointing to an allowed URI to prevent
unauthorized remote uri invocation. Validates only
+ * the URI relevant to the callback type against the corresponding
allowlist.
*/
- private void verifyRequest(Exchange exchange) {
+ private void verifyRequest(Exchange exchange, boolean isCompensation) {
if (exchange.getIn().getHeader(Exchange.SAGA_LONG_RUNNING_ACTION) ==
null) {
throw new IllegalArgumentException("Missing " +
Exchange.SAGA_LONG_RUNNING_ACTION + " header in received request");
}
- Set<String> usedURIs = new HashSet<>();
String compensationURI =
exchange.getIn().getHeader(URL_COMPENSATION_KEY, String.class);
- if (compensationURI != null) {
- usedURIs.add(compensationURI);
- }
String completionURI = exchange.getIn().getHeader(URL_COMPLETION_KEY,
String.class);
- if (completionURI != null) {
- usedURIs.add(completionURI);
- }
// CAMEL-17751: Extract URIs from the CamelHttpQuery header
- if (usedURIs.isEmpty()) {
+ if (compensationURI == null && completionURI == null) {
Map<String, String> queryParams
=
parseQuery(exchange.getIn().getHeader(Exchange.HTTP_QUERY, String.class));
if (!queryParams.isEmpty()) {
-
if (queryParams.get(URL_COMPENSATION_KEY) != null) {
compensationURI = queryParams.get(URL_COMPENSATION_KEY);
- usedURIs.add(compensationURI);
exchange.getIn().setHeader(URL_COMPENSATION_KEY,
compensationURI);
}
-
if (queryParams.get(URL_COMPLETION_KEY) != null) {
completionURI = queryParams.get(URL_COMPLETION_KEY);
- usedURIs.add(completionURI);
exchange.getIn().setHeader(URL_COMPLETION_KEY,
completionURI);
}
}
}
- for (String uri : usedURIs) {
- if (!sagaService.getRegisteredURIs().contains(uri)) {
+ String uri = isCompensation ? compensationURI : completionURI;
+ if (uri != null) {
+ Set<String> allowedURIs = isCompensation
+ ? sagaService.getRegisteredCompensationURIs()
+ : sagaService.getRegisteredCompletionURIs();
+ if (!allowedURIs.contains(uri)) {
throw new IllegalArgumentException("URI " + uri + " is not
allowed");
}
}
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 73552c994ffb..c9c4f7f208a6 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
@@ -17,9 +17,13 @@
package org.apache.camel.service.lra;
import java.net.URI;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ScheduledExecutorService;
import org.apache.camel.*;
@@ -29,8 +33,10 @@ import org.apache.camel.saga.CamelSagaCoordinator;
import org.apache.camel.saga.CamelSagaService;
import org.apache.camel.saga.CamelSagaStep;
import org.apache.camel.spi.Configurer;
+import org.apache.camel.spi.LifecycleStrategy;
import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.annotations.JdkService;
+import org.apache.camel.support.LifecycleStrategySupport;
import org.apache.camel.support.service.ServiceSupport;
/**
@@ -45,7 +51,10 @@ public class LRASagaService extends ServiceSupport
implements StaticService, Cam
private ScheduledExecutorService executorService;
private LRAClient client;
private LRASagaRoutes routes;
- private final Set<String> sagaURIs = ConcurrentHashMap.newKeySet();
+ private LifecycleStrategy lifecycleStrategy;
+ private final Set<String> compensationURIs = ConcurrentHashMap.newKeySet();
+ private final Set<String> completionURIs = ConcurrentHashMap.newKeySet();
+ private final ConcurrentHashMap<String, List<CamelSagaStep>>
stepsByRouteId = new ConcurrentHashMap<>();
// we want to be able to configure these following options
@Metadata
@@ -79,8 +88,38 @@ public class LRASagaService extends ServiceSupport
implements StaticService, Cam
@Override
public void registerStep(CamelSagaStep step) {
-
step.getCompensation().map(Endpoint::getEndpointUri).ifPresent(this.sagaURIs::add);
-
step.getCompletion().map(Endpoint::getEndpointUri).ifPresent(this.sagaURIs::add);
+
step.getCompensation().map(Endpoint::getEndpointUri).ifPresent(compensationURIs::add);
+
step.getCompletion().map(Endpoint::getEndpointUri).ifPresent(completionURIs::add);
+ if (step.getRouteId() != null) {
+ stepsByRouteId.computeIfAbsent(step.getRouteId(), k -> new
CopyOnWriteArrayList<>()).add(step);
+ }
+ }
+
+ void unregisterSteps(String routeId) {
+ List<CamelSagaStep> steps = stepsByRouteId.remove(routeId);
+ if (steps == null) {
+ return;
+ }
+
+ // collect URIs still referenced by other routes
+ Set<String> retainedCompensation = new HashSet<>();
+ Set<String> retainedCompletion = new HashSet<>();
+ for (List<CamelSagaStep> remaining : stepsByRouteId.values()) {
+ for (CamelSagaStep s : remaining) {
+
s.getCompensation().map(Endpoint::getEndpointUri).ifPresent(retainedCompensation::add);
+
s.getCompletion().map(Endpoint::getEndpointUri).ifPresent(retainedCompletion::add);
+ }
+ }
+
+ // remove only URIs no longer referenced
+ for (CamelSagaStep step : steps) {
+ step.getCompensation().map(Endpoint::getEndpointUri)
+ .filter(uri -> !retainedCompensation.contains(uri))
+ .ifPresent(compensationURIs::remove);
+ step.getCompletion().map(Endpoint::getEndpointUri)
+ .filter(uri -> !retainedCompletion.contains(uri))
+ .ifPresent(completionURIs::remove);
+ }
}
@Override
@@ -121,6 +160,13 @@ public class LRASagaService extends ServiceSupport
implements StaticService, Cam
this.client.close();
this.client = null;
}
+ if (this.lifecycleStrategy != null) {
+
camelContext.getLifecycleStrategies().remove(this.lifecycleStrategy);
+ this.lifecycleStrategy = null;
+ }
+ compensationURIs.clear();
+ completionURIs.clear();
+ stepsByRouteId.clear();
}
@Override
@@ -136,6 +182,17 @@ public class LRASagaService extends ServiceSupport
implements StaticService, Cam
throw RuntimeCamelException.wrapRuntimeException(ex);
}
}
+ if (this.lifecycleStrategy == null) {
+ this.lifecycleStrategy = new LifecycleStrategySupport() {
+ @Override
+ public void onRoutesRemove(Collection<Route> routes) {
+ for (Route r : routes) {
+ unregisterSteps(r.getRouteId());
+ }
+ }
+ };
+ this.camelContext.addLifecycleStrategy(this.lifecycleStrategy);
+ }
}
@Override
@@ -187,8 +244,18 @@ public class LRASagaService extends ServiceSupport
implements StaticService, Cam
this.localParticipantContextPath = localParticipantContextPath;
}
+ public Set<String> getRegisteredCompensationURIs() {
+ return compensationURIs;
+ }
+
+ public Set<String> getRegisteredCompletionURIs() {
+ return completionURIs;
+ }
+
public Set<String> getRegisteredURIs() {
- return sagaURIs;
+ Set<String> all = new HashSet<>(compensationURIs);
+ all.addAll(completionURIs);
+ return all;
}
@Override
diff --git
a/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRASagaServiceAllowlistTest.java
b/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRASagaServiceAllowlistTest.java
new file mode 100644
index 000000000000..905ed19416b6
--- /dev/null
+++
b/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRASagaServiceAllowlistTest.java
@@ -0,0 +1,148 @@
+/*
+ * 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.service.lra;
+
+import java.util.Collections;
+import java.util.Set;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.saga.CamelSagaStep;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+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;
+
+public class LRASagaServiceAllowlistTest extends CamelTestSupport {
+
+ private LRASagaService sagaService;
+
+ public LRASagaServiceAllowlistTest() {
+ testConfiguration().withUseRouteBuilder(false);
+ }
+
+ @BeforeEach
+ void initService() {
+ sagaService = new LRASagaService();
+ sagaService.setCoordinatorUrl("http://localhost:8080";);
+ sagaService.setLocalParticipantUrl("http://localhost:8081";);
+ sagaService.setLocalParticipantContextPath("/lra-participant");
+ sagaService.setCoordinatorContextPath("/lra-coordinator");
+ sagaService.setCamelContext(context());
+ }
+
+ @DisplayName("registerStep adds compensation URI only to the compensation
allowlist")
+ @Test
+ void testCompensationUriSeparation() {
+ Endpoint compensation = context().getEndpoint("direct:compensate1");
+ CamelSagaStep step = new CamelSagaStep(compensation, null,
Collections.emptyMap(), null);
+ step.setRouteId("route1");
+
+ sagaService.registerStep(step);
+
+
assertTrue(sagaService.getRegisteredCompensationURIs().contains("direct://compensate1"));
+
assertFalse(sagaService.getRegisteredCompletionURIs().contains("direct://compensate1"));
+ }
+
+ @DisplayName("registerStep adds completion URI only to the completion
allowlist")
+ @Test
+ void testCompletionUriSeparation() {
+ Endpoint completion = context().getEndpoint("direct:complete1");
+ CamelSagaStep step = new CamelSagaStep(null, completion,
Collections.emptyMap(), null);
+ step.setRouteId("route1");
+
+ sagaService.registerStep(step);
+
+
assertTrue(sagaService.getRegisteredCompletionURIs().contains("direct://complete1"));
+
assertFalse(sagaService.getRegisteredCompensationURIs().contains("direct://complete1"));
+ }
+
+ @DisplayName("getRegisteredURIs returns the union of both sets")
+ @Test
+ void testGetRegisteredURIsReturnsUnion() {
+ Endpoint compensation = context().getEndpoint("direct:compensate1");
+ Endpoint completion = context().getEndpoint("direct:complete1");
+ CamelSagaStep step = new CamelSagaStep(compensation, completion,
Collections.emptyMap(), null);
+ step.setRouteId("route1");
+
+ sagaService.registerStep(step);
+
+ Set<String> all = sagaService.getRegisteredURIs();
+ assertEquals(2, all.size());
+ assertTrue(all.contains("direct://compensate1"));
+ assertTrue(all.contains("direct://complete1"));
+ }
+
+ @DisplayName("unregisterSteps prunes URIs from the allowlists")
+ @Test
+ void testUnregisterStepsPrunesURIs() {
+ Endpoint compensation = context().getEndpoint("direct:compensate1");
+ Endpoint completion = context().getEndpoint("direct:complete1");
+ CamelSagaStep step = new CamelSagaStep(compensation, completion,
Collections.emptyMap(), null);
+ step.setRouteId("testRoute");
+
+ sagaService.registerStep(step);
+
assertTrue(sagaService.getRegisteredCompensationURIs().contains("direct://compensate1"));
+
assertTrue(sagaService.getRegisteredCompletionURIs().contains("direct://complete1"));
+
+ sagaService.unregisterSteps("testRoute");
+
+
assertFalse(sagaService.getRegisteredCompensationURIs().contains("direct://compensate1"));
+
assertFalse(sagaService.getRegisteredCompletionURIs().contains("direct://complete1"));
+ assertTrue(sagaService.getRegisteredURIs().isEmpty());
+ }
+
+ @DisplayName("Shared URIs are only pruned when all referencing routes are
removed")
+ @Test
+ void testSharedUriRetainedUntilAllRoutesRemoved() {
+ Endpoint sharedCompensation =
context().getEndpoint("direct:sharedCompensate");
+
+ CamelSagaStep step1 = new CamelSagaStep(sharedCompensation, null,
Collections.emptyMap(), null);
+ step1.setRouteId("routeA");
+ sagaService.registerStep(step1);
+
+ CamelSagaStep step2 = new CamelSagaStep(sharedCompensation, null,
Collections.emptyMap(), null);
+ step2.setRouteId("routeB");
+ sagaService.registerStep(step2);
+
+
assertTrue(sagaService.getRegisteredCompensationURIs().contains("direct://sharedCompensate"));
+
+ // remove routeA — shared URI should remain because routeB still
references it
+ sagaService.unregisterSteps("routeA");
+
assertTrue(sagaService.getRegisteredCompensationURIs().contains("direct://sharedCompensate"));
+
+ // remove routeB — now the URI should be pruned
+ sagaService.unregisterSteps("routeB");
+
assertFalse(sagaService.getRegisteredCompensationURIs().contains("direct://sharedCompensate"));
+ }
+
+ @DisplayName("unregisterSteps for unknown routeId is a no-op")
+ @Test
+ void testUnregisterUnknownRouteIsNoOp() {
+ Endpoint compensation = context().getEndpoint("direct:compensate1");
+ CamelSagaStep step = new CamelSagaStep(compensation, null,
Collections.emptyMap(), null);
+ step.setRouteId("route1");
+ sagaService.registerStep(step);
+
+ sagaService.unregisterSteps("unknownRoute");
+
+
assertTrue(sagaService.getRegisteredCompensationURIs().contains("direct://compensate1"));
+ }
+}
diff --git
a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SagaReifier.java
b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SagaReifier.java
index 2a3568b1ce9c..37d284c1e9ff 100644
---
a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SagaReifier.java
+++
b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SagaReifier.java
@@ -95,6 +95,7 @@ public class SagaReifier extends
ProcessorReifier<SagaDefinition> {
CamelSagaStep step = new CamelSagaStep(
compensationEndpoint, completionEndpoint, optionsMap,
parseDuration(timeout));
+ step.setRouteId(route.getRouteId());
SagaPropagation propagation = parse(SagaPropagation.class,
definition.getPropagation());
if (propagation == null) {
diff --git
a/core/camel-support/src/main/java/org/apache/camel/saga/CamelSagaStep.java
b/core/camel-support/src/main/java/org/apache/camel/saga/CamelSagaStep.java
index 2fdd83847bd8..1b2a57091929 100644
--- a/core/camel-support/src/main/java/org/apache/camel/saga/CamelSagaStep.java
+++ b/core/camel-support/src/main/java/org/apache/camel/saga/CamelSagaStep.java
@@ -35,6 +35,8 @@ public class CamelSagaStep {
private final Long timeoutInMilliseconds;
+ private String routeId;
+
public CamelSagaStep(Endpoint compensation, Endpoint completion,
Map<String, Expression> options, Long
timeoutInMilliseconds) {
this.compensation = compensation;
@@ -59,6 +61,14 @@ public class CamelSagaStep {
return Optional.ofNullable(timeoutInMilliseconds);
}
+ public String getRouteId() {
+ return routeId;
+ }
+
+ public void setRouteId(String routeId) {
+ this.routeId = routeId;
+ }
+
public boolean isEmpty() {
return compensation == null && completion == null && options.isEmpty()
&& timeoutInMilliseconds == null;
}
