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 d7e36c21b587 CAMEL-25135: routeConfiguration onCompletion is skipped
in the Java DSL (#27430)
d7e36c21b587 is described below
commit d7e36c21b587a6c3c9557a1f3cf944398971db24
Author: Guillaume Nodet <[email protected]>
AuthorDate: Thu Oct 8 10:45:46 2026 +0200
CAMEL-25135: routeConfiguration onCompletion is skipped in the Java DSL
(#27430)
Co-authored-by: Claude Sonnet 4.6 <[email protected]>
---
.../org/apache/camel/ExchangeConstantProvider.java | 4 +-
.../src/main/java/org/apache/camel/Exchange.java | 11 +
.../java/org/apache/camel/ExchangePropertyKey.java | 6 +
.../camel/model/RouteConfigurationDefinition.java | 14 +-
.../camel/processor/OnCompletionProcessor.java | 114 +++++++-
.../apache/camel/reifier/OnCompletionReifier.java | 10 +-
.../model/RoutesConfigurationOnCompletionTest.java | 302 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 15 +
8 files changed, 468 insertions(+), 8 deletions(-)
diff --git
a/core/camel-api/src/generated/java/org/apache/camel/ExchangeConstantProvider.java
b/core/camel-api/src/generated/java/org/apache/camel/ExchangeConstantProvider.java
index a9d27b5a2870..2b78c48f937c 100644
---
a/core/camel-api/src/generated/java/org/apache/camel/ExchangeConstantProvider.java
+++
b/core/camel-api/src/generated/java/org/apache/camel/ExchangeConstantProvider.java
@@ -17,7 +17,7 @@ public class ExchangeConstantProvider {
private static final Map<String, String> MAP;
static {
- Map<String, String> map = new HashMap<>(158);
+ Map<String, String> map = new HashMap<>(160);
map.put("ACTIVITY_SPAN_TAGS", "CamelActivitySpanTags");
map.put("AGGREGATED_COLLECTION_GUARD",
"CamelAggregatedCollectionGuard");
map.put("AGGREGATED_COMPLETED_BY", "CamelAggregatedCompletedBy");
@@ -130,6 +130,8 @@ public class ExchangeConstantProvider {
map.put("MULTICAST_INDEX", "CamelMulticastIndex");
map.put("OFFSET", "CamelOffset");
map.put("ON_COMPLETION", "CamelOnCompletion");
+ map.put("ON_COMPLETION_CONFIG_REGISTERED_ROUTES",
"CamelOnCompletionConfigRegisteredRoutes");
+ map.put("ON_COMPLETION_FIRED_IDS", "CamelOnCompletionFiredIds");
map.put("ON_COMPLETION_ROUTE_IDS", "CamelOnCompletionRouteIds");
map.put("OTEL_ACTIVE_SPAN", "OpenTracing.activeSpan");
map.put("OTEL_CLOSE_CLIENT_SCOPE", "OpenTracing.closeClientScope");
diff --git a/core/camel-api/src/main/java/org/apache/camel/Exchange.java
b/core/camel-api/src/main/java/org/apache/camel/Exchange.java
index 726ba1689523..8b9aa9bc7a2d 100644
--- a/core/camel-api/src/main/java/org/apache/camel/Exchange.java
+++ b/core/camel-api/src/main/java/org/apache/camel/Exchange.java
@@ -255,6 +255,17 @@ public interface Exchange extends VariableAware {
description = "Flag to mark that this exchange is currently
being executed as onCompletion", javaType = "boolean")
String ON_COMPLETION = "CamelOnCompletion";
String ON_COMPLETION_ROUTE_IDS = "CamelOnCompletionRouteIds";
+ /**
+ * Exchange property key for tracking which OnCompletionProcessor
instances (by processor id) have already fired
+ * their onCompletion handler for this exchange. Used to ensure a named
route configuration's onCompletion fires at
+ * most once per exchange, even when multiple opted-in routes are
traversed in the same exchange.
+ */
+ String ON_COMPLETION_FIRED_IDS = "CamelOnCompletionFiredIds";
+ /**
+ * Exchange property key for tracking which routes have registered a
synchronization for each named route
+ * configuration ID. Used to defer BeforeConsumer firing from an inner
route when the consumer route also opted in.
+ */
+ String ON_COMPLETION_CONFIG_REGISTERED_ROUTES =
"CamelOnCompletionConfigRegisteredRoutes";
String OFFSET = "CamelOffset";
String OVERRULE_FILE_NAME = "CamelOverruleFileName";
diff --git
a/core/camel-api/src/main/java/org/apache/camel/ExchangePropertyKey.java
b/core/camel-api/src/main/java/org/apache/camel/ExchangePropertyKey.java
index 8a90ce54e5d7..fa779db84212 100644
--- a/core/camel-api/src/main/java/org/apache/camel/ExchangePropertyKey.java
+++ b/core/camel-api/src/main/java/org/apache/camel/ExchangePropertyKey.java
@@ -81,6 +81,8 @@ public enum ExchangePropertyKey {
MULTICAST_INDEX(Exchange.MULTICAST_INDEX),
ON_COMPLETION(Exchange.ON_COMPLETION),
ON_COMPLETION_ROUTE_IDS(Exchange.ON_COMPLETION_ROUTE_IDS),
+ ON_COMPLETION_FIRED_IDS(Exchange.ON_COMPLETION_FIRED_IDS),
+
ON_COMPLETION_CONFIG_REGISTERED_ROUTES(Exchange.ON_COMPLETION_CONFIG_REGISTERED_ROUTES),
@Deprecated(since = "4.14.0")
PARENT_UNIT_OF_WORK(Exchange.PARENT_UNIT_OF_WORK),
RECEIVED_TIMESTAMP(Exchange.RECEIVED_TIMESTAMP),
@@ -215,6 +217,10 @@ public enum ExchangePropertyKey {
return ON_COMPLETION;
case Exchange.ON_COMPLETION_ROUTE_IDS:
return ON_COMPLETION_ROUTE_IDS;
+ case Exchange.ON_COMPLETION_FIRED_IDS:
+ return ON_COMPLETION_FIRED_IDS;
+ case Exchange.ON_COMPLETION_CONFIG_REGISTERED_ROUTES:
+ return ON_COMPLETION_CONFIG_REGISTERED_ROUTES;
case Exchange.PARENT_UNIT_OF_WORK:
return PARENT_UNIT_OF_WORK;
case Exchange.RECEIVED_TIMESTAMP:
diff --git
a/core/camel-core-model/src/main/java/org/apache/camel/model/RouteConfigurationDefinition.java
b/core/camel-core-model/src/main/java/org/apache/camel/model/RouteConfigurationDefinition.java
index e235f79a66c2..7e83851855bb 100644
---
a/core/camel-core-model/src/main/java/org/apache/camel/model/RouteConfigurationDefinition.java
+++
b/core/camel-core-model/src/main/java/org/apache/camel/model/RouteConfigurationDefinition.java
@@ -233,8 +233,18 @@ public class RouteConfigurationDefinition extends
OptionalIdentifiedDefinition<R
public OnCompletionDefinition onCompletion() {
OnCompletionDefinition answer = new OnCompletionDefinition();
answer.setRouteConfiguration(this);
- // is global scoped by default
- answer.setRouteScoped(false);
+ if (getId() == null || "*".equals(getId())) {
+ // route configuration with no id (or wildcard id "*") applies
globally to all routes,
+ // so the onCompletion must not be route-scoped: the de-duplication
+ // in OnCompletionProcessor.shouldSkip() relies on
routeScoped=false
+ // to fire only once per exchange (for the consumer route).
+ answer.setRouteScoped(false);
+ }
+ // named route configurations keep routeScoped=true (the model
default) so that
+ // OnCompletionProcessor uses the route-visit tracking mechanism and
fires correctly
+ // even when the opted-in route is called indirectly from a consumer
route
+ // (e.g. REST DSL → direct:). A per-configuration dedup key prevents
double-firing
+ // when multiple routes share the same named configuration.
onCompletions.add(answer);
return answer;
}
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/OnCompletionProcessor.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/OnCompletionProcessor.java
index 63fb2f3b01a8..519996181c8f 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/OnCompletionProcessor.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/OnCompletionProcessor.java
@@ -18,7 +18,10 @@ package org.apache.camel.processor;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
@@ -73,6 +76,9 @@ public class OnCompletionProcessor extends
BaseProcessorSupport
private final boolean useOriginalBody;
private final boolean afterConsumer;
private final boolean routeScoped;
+ // non-null only for named (route-scoped) route configurations; used to
dedup
+ // across multiple OnCompletionProcessor instances sharing the same
configuration
+ private final String configurationId;
private final LongAdder taskCount = new LongAdder();
// the parallel onCompletion tasks that have been submitted but have not
started yet
private final Set<ParallelTask> pendingTasks =
ConcurrentHashMap.newKeySet();
@@ -81,6 +87,14 @@ public class OnCompletionProcessor extends
BaseProcessorSupport
boolean shutdownExecutorService,
boolean onCompleteOnly, boolean
onFailureOnly, Predicate onWhen, boolean useOriginalBody,
boolean afterConsumer, boolean routeScoped) {
+ this(camelContext, processor, executorService, shutdownExecutorService,
+ onCompleteOnly, onFailureOnly, onWhen, useOriginalBody,
afterConsumer, routeScoped, null);
+ }
+
+ public OnCompletionProcessor(CamelContext camelContext, Processor
processor, ExecutorService executorService,
+ boolean shutdownExecutorService,
+ boolean onCompleteOnly, boolean
onFailureOnly, Predicate onWhen, boolean useOriginalBody,
+ boolean afterConsumer, boolean routeScoped,
String configurationId) {
notNull(camelContext, "camelContext");
notNull(processor, "processor");
this.camelContext = camelContext;
@@ -93,6 +107,7 @@ public class OnCompletionProcessor extends
BaseProcessorSupport
this.useOriginalBody = useOriginalBody;
this.afterConsumer = afterConsumer;
this.routeScoped = routeScoped;
+ this.configurationId = configurationId;
}
@Override
@@ -182,13 +197,29 @@ public class OnCompletionProcessor extends
BaseProcessorSupport
@Override
public boolean process(Exchange exchange, AsyncCallback callback) {
if (processor != null) {
+ // For named route configurations with BeforeConsumer mode,
register the route that owns this
+ // processor so that BeforeConsumer can determine whether to defer
to the consumer route.
+ // (AfterConsumer does not use this map, so we skip the allocation
to keep the hot path clean.)
+ if (!afterConsumer && configurationId != null) {
+ @SuppressWarnings("unchecked")
+ Map<String, Set<String>> registeredRoutes
+ =
exchange.getProperty(ExchangePropertyKey.ON_COMPLETION_CONFIG_REGISTERED_ROUTES,
Map.class);
+ if (registeredRoutes == null) {
+ registeredRoutes = new HashMap<>();
+
exchange.setProperty(ExchangePropertyKey.ON_COMPLETION_CONFIG_REGISTERED_ROUTES,
registeredRoutes);
+ }
+ registeredRoutes.computeIfAbsent(configurationId, k -> new
HashSet<>()).add(getRouteId());
+ }
+
// register callback
if (afterConsumer) {
exchange.getUnitOfWork()
- .addSynchronization(new
OnCompletionSynchronizationAfterConsumer(routeScoped, getRouteId()));
+ .addSynchronization(
+ new
OnCompletionSynchronizationAfterConsumer(routeScoped, getRouteId(),
configurationId));
} else {
exchange.getUnitOfWork()
- .addSynchronization(new
OnCompletionSynchronizationBeforeConsumer(routeScoped, getRouteId()));
+ .addSynchronization(
+ new
OnCompletionSynchronizationBeforeConsumer(routeScoped, getRouteId(),
configurationId));
}
}
@@ -401,10 +432,12 @@ public class OnCompletionProcessor extends
BaseProcessorSupport
private final boolean routeScoped;
private final String routeId;
+ private final String configurationId;
- public OnCompletionSynchronizationAfterConsumer(boolean routeScoped,
String routeId) {
+ public OnCompletionSynchronizationAfterConsumer(boolean routeScoped,
String routeId, String configurationId) {
this.routeScoped = routeScoped;
this.routeId = routeId;
+ this.configurationId = configurationId;
}
@Override
@@ -443,6 +476,9 @@ public class OnCompletionProcessor extends
BaseProcessorSupport
return;
}
+ // record dedup AFTER all checks pass, using processor id as key
+ recordFired(exchange);
+
// must use a copy as we don't want it to cause side effects of
the original exchange
final Exchange copy = prepareExchange(exchange);
@@ -465,6 +501,9 @@ public class OnCompletionProcessor extends
BaseProcessorSupport
return;
}
+ // record dedup AFTER all checks pass, using processor id as key
+ recordFired(exchange);
+
// must use a copy as we don't want it to cause side effects of
the original exchange
final Exchange copy = prepareExchange(exchange);
final Exception original = copy.getException();
@@ -504,6 +543,16 @@ public class OnCompletionProcessor extends
BaseProcessorSupport
if (routeIds == null || !routeIds.contains(routeId)) {
return true;
}
+ // named configuration: fire only once per exchange per
OnCompletionDefinition.
+ // Use the processor id (unique per definition) as the dedup
key so that
+ // onCompleteOnly and onFailureOnly blocks in the same config
dedup independently.
+ if (configurationId != null) {
+ Set<String> firedIds
+ =
exchange.getProperty(ExchangePropertyKey.ON_COMPLETION_FIRED_IDS, Set.class);
+ if (firedIds != null &&
firedIds.contains(OnCompletionProcessor.this.id)) {
+ return true;
+ }
+ }
}
if (onCompleteOrOnFailureOnly) {
@@ -518,6 +567,19 @@ public class OnCompletionProcessor extends
BaseProcessorSupport
return false;
}
+ @SuppressWarnings("unchecked")
+ private void recordFired(Exchange exchange) {
+ if (configurationId != null) {
+ Set<String> firedIds
+ =
exchange.getProperty(ExchangePropertyKey.ON_COMPLETION_FIRED_IDS, Set.class);
+ if (firedIds == null) {
+ firedIds = new HashSet<>();
+
exchange.setProperty(ExchangePropertyKey.ON_COMPLETION_FIRED_IDS, firedIds);
+ }
+ firedIds.add(OnCompletionProcessor.this.id);
+ }
+ }
+
@Override
public String toString() {
if (!onCompleteOnly && !onFailureOnly) {
@@ -554,10 +616,12 @@ public class OnCompletionProcessor extends
BaseProcessorSupport
private final boolean routeScoped;
private final String routeId;
+ private final String configurationId;
- public OnCompletionSynchronizationBeforeConsumer(boolean routeScoped,
String routeId) {
+ public OnCompletionSynchronizationBeforeConsumer(boolean routeScoped,
String routeId, String configurationId) {
this.routeScoped = routeScoped;
this.routeId = routeId;
+ this.configurationId = configurationId;
}
@Override
@@ -587,6 +651,36 @@ public class OnCompletionProcessor extends
BaseProcessorSupport
return;
}
+ // named configuration: defer firing to the consumer route
when the consumer
+ // also opted in to the same named configuration (so
onCompletion sees state
+ // set after the sub-route call).
+ if (routeScoped && configurationId != null) {
+ String fromRouteId = exchange.getFromRouteId();
+ if (fromRouteId != null &&
!routeId.equals(fromRouteId)) {
+ @SuppressWarnings("unchecked")
+ Map<String, Set<String>> registeredRoutes =
exchange.getProperty(
+
ExchangePropertyKey.ON_COMPLETION_CONFIG_REGISTERED_ROUTES, Map.class);
+ if (registeredRoutes != null) {
+ Set<String> routes =
registeredRoutes.get(configurationId);
+ if (routes != null &&
routes.contains(fromRouteId)) {
+ // consumer route also opted in; defer to
its onAfterRoute
+ return;
+ }
+ }
+ }
+ }
+
+ // named configuration: fire only once per exchange per
OnCompletionDefinition.
+ // Use the processor id (unique per definition) as the
dedup key.
+ if (routeScoped && configurationId != null) {
+ @SuppressWarnings("unchecked")
+ Set<String> firedIds
+ =
exchange.getProperty(ExchangePropertyKey.ON_COMPLETION_FIRED_IDS, Set.class);
+ if (firedIds != null &&
firedIds.contains(OnCompletionProcessor.this.id)) {
+ return; // already fired for this definition
+ }
+ }
+
if (exchange.isFailed() && onCompleteOnly) {
return;
}
@@ -600,6 +694,18 @@ public class OnCompletionProcessor extends
BaseProcessorSupport
return;
}
+ // record dedup AFTER all checks pass
+ if (configurationId != null) {
+ @SuppressWarnings("unchecked")
+ Set<String> firedIds
+ =
exchange.getProperty(ExchangePropertyKey.ON_COMPLETION_FIRED_IDS, Set.class);
+ if (firedIds == null) {
+ firedIds = new HashSet<>();
+
exchange.setProperty(ExchangePropertyKey.ON_COMPLETION_FIRED_IDS, firedIds);
+ }
+ firedIds.add(OnCompletionProcessor.this.id);
+ }
+
// must use a copy as we don't want it to cause side
effects of the original exchange
final Exchange copy = prepareExchange(exchange);
diff --git
a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/OnCompletionReifier.java
b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/OnCompletionReifier.java
index efef06580e10..5de075f4ddf2 100644
---
a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/OnCompletionReifier.java
+++
b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/OnCompletionReifier.java
@@ -71,9 +71,17 @@ public class OnCompletionReifier extends
ProcessorReifier<OnCompletionDefinition
boolean afterConsumer = definition.getMode() == null
|| parse(OnCompletionMode.class, definition.getMode()) ==
OnCompletionMode.AfterConsumer;
+ // Determine the route configuration id for named (route-scoped)
onCompletion dedup.
+ // When multiple routes share the same named configuration, each gets
its own
+ // OnCompletionProcessor, but they should fire only once per exchange
total.
+ String configurationId = null;
+ if (definition.isRouteScoped() && definition.getRouteConfiguration()
!= null) {
+ configurationId = definition.getRouteConfiguration().getId();
+ }
+
OnCompletionProcessor answer = new OnCompletionProcessor(
camelContext, target, threadPool, shutdownThreadPool,
isOnCompleteOnly, isOnFailureOnly, when,
- original, afterConsumer, definition.isRouteScoped());
+ original, afterConsumer, definition.isRouteScoped(),
configurationId);
answer.setDisabled(isDisabled(camelContext, definition));
return answer;
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/model/RoutesConfigurationOnCompletionTest.java
b/core/camel-core/src/test/java/org/apache/camel/model/RoutesConfigurationOnCompletionTest.java
index 6a0ff4b96acf..38c5308b64bb 100644
---
a/core/camel-core/src/test/java/org/apache/camel/model/RoutesConfigurationOnCompletionTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/model/RoutesConfigurationOnCompletionTest.java
@@ -16,11 +16,18 @@
*/
package org.apache.camel.model;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.builder.RouteConfigurationBuilder;
import org.junit.jupiter.api.Test;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
public class RoutesConfigurationOnCompletionTest extends ContextTestSupport {
@Override
@@ -122,4 +129,299 @@ public class RoutesConfigurationOnCompletionTest extends
ContextTestSupport {
assertMockEndpointsSatisfied();
}
+ @Test
+ public void testLocalConfigurationWithIndirectConsumer() throws Exception {
+ // CAMEL-25135: onCompletion from a named routeConfiguration must fire
even when
+ // the opted-in route is not itself the consumer route (e.g. REST DSL
→ direct:).
+ context.addRoutes(new RouteConfigurationBuilder() {
+ @Override
+ public void configuration() {
+
routeConfiguration("myconfig").onCompletion().to("mock:completion");
+ }
+ });
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ // consumer route — does NOT carry the routeConfigurationId
+ from("direct:consumer")
+ .to("direct:processor");
+
+ // processing route — opts in to the named configuration
+ from("direct:processor").routeConfigurationId("myconfig")
+ .to("mock:result");
+ }
+ });
+ context.start();
+
+ getMockEndpoint("mock:completion").expectedMessageCount(1);
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+
+ template.sendBody("direct:consumer", "Hello World");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testLocalConfigurationBothRoutesOptIn() throws Exception {
+ // CAMEL-25135: when both the consumer route and the called direct:
route
+ // opt in to the same named configuration, onCompletion must fire only
once
+ // per exchange, not once per opted-in route.
+ context.addRoutes(new RouteConfigurationBuilder() {
+ @Override
+ public void configuration() {
+
routeConfiguration("myconfig").onCompletion().to("mock:completion");
+ }
+ });
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ // both routes opt in to the same named configuration
+ from("direct:consumer").routeConfigurationId("myconfig")
+ .to("direct:processor");
+
+ from("direct:processor").routeConfigurationId("myconfig")
+ .to("mock:result");
+ }
+ });
+ context.start();
+
+ getMockEndpoint("mock:completion").expectedMessageCount(1);
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+
+ template.sendBody("direct:consumer", "Hello World");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testLocalConfigurationSubRouteCalledTwice() throws Exception {
+ // CAMEL-25135: when the opted-in processor route is called twice in
the same
+ // exchange, onCompletion must fire only once per exchange, not once
per visit.
+ context.addRoutes(new RouteConfigurationBuilder() {
+ @Override
+ public void configuration() {
+
routeConfiguration("myconfig").onCompletion().to("mock:completion");
+ }
+ });
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:consumer")
+ .to("direct:processor")
+ .to("direct:processor"); // called twice
+
+ from("direct:processor").routeConfigurationId("myconfig")
+ .to("mock:result");
+ }
+ });
+ context.start();
+
+ getMockEndpoint("mock:completion").expectedMessageCount(1);
+ getMockEndpoint("mock:result").expectedMessageCount(2); // called twice
+
+ template.sendBody("direct:consumer", "Hello World");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testLocalConfigurationBeforeConsumerWithIndirectConsumer()
throws Exception {
+ // CAMEL-25135: BeforeConsumer mode with a named routeConfiguration
must fire
+ // when the opted-in route is not itself the consumer route.
+ context.addRoutes(new RouteConfigurationBuilder() {
+ @Override
+ public void configuration() {
+
routeConfiguration("myconfig").onCompletion().modeBeforeConsumer()
+ .setHeader("done", constant("yes"));
+ }
+ });
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ // consumer route — does NOT carry the routeConfigurationId
+ from("direct:consumer")
+ .to("direct:processor");
+
+ // processing route — opts in to the named configuration
+ from("direct:processor").routeConfigurationId("myconfig")
+ .to("mock:result");
+ }
+ });
+ context.start();
+
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+
+ Exchange result = template.request("direct:consumer",
+ e -> e.getMessage().setBody("Hello World"));
+
+ assertMockEndpointsSatisfied();
+ // BeforeConsumer onCompletion runs before the consumer sends the
reply,
+ // so the header should be visible on the reply exchange
+ assertEquals("yes", result.getMessage().getHeader("done"));
+ }
+
+ @Test
+ public void testLocalConfigurationBeforeConsumerBothRoutesOptIn() throws
Exception {
+ // CAMEL-25135: BeforeConsumer + both routes opt in → fire only once.
+ context.addRoutes(new RouteConfigurationBuilder() {
+ @Override
+ public void configuration() {
+
routeConfiguration("myconfig").onCompletion().modeBeforeConsumer()
+ .setHeader("done", constant("yes"));
+ }
+ });
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:consumer").routeConfigurationId("myconfig")
+ .to("direct:processor");
+
+ from("direct:processor").routeConfigurationId("myconfig")
+ .to("mock:result");
+ }
+ });
+ context.start();
+
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+
+ Exchange result = template.request("direct:consumer",
+ e -> e.getMessage().setBody("Hello World"));
+
+ assertMockEndpointsSatisfied();
+ assertEquals("yes", result.getMessage().getHeader("done"));
+ }
+
+ @Test
+ public void testGlobalStarConfiguration() throws Exception {
+ // routeConfiguration("*") is treated as global: fires once per
exchange for all routes
+ context.addRoutes(new RouteConfigurationBuilder() {
+ @Override
+ public void configuration() {
+ routeConfiguration("*").onCompletion().to("mock:global");
+ }
+ });
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .to("mock:result");
+
+ from("direct:start2")
+ .to("mock:result2");
+
+ // multi-route exchange: start3 calls sub; must fire
mock:global exactly once
+ from("direct:start3")
+ .to("direct:sub");
+
+ from("direct:sub")
+ .to("mock:sub");
+ }
+ });
+ context.start();
+
+ getMockEndpoint("mock:global").expectedMessageCount(3); // one per
exchange, not per route
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+ getMockEndpoint("mock:result2").expectedBodiesReceived("Bye World");
+ getMockEndpoint("mock:sub").expectedBodiesReceived("Two-Hop World");
+
+ template.sendBody("direct:start", "Hello World");
+ template.sendBody("direct:start2", "Bye World");
+ template.sendBody("direct:start3", "Two-Hop World");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testOnCompleteOnlyAndOnFailureOnlyInSameNamedConfig() throws
Exception {
+ // One named config with onCompleteOnly + onFailureOnly: they must not
dedup-block each other.
+ context.addRoutes(new RouteConfigurationBuilder() {
+ @Override
+ public void configuration() {
+ RouteConfigurationDefinition rc =
routeConfiguration("myconfig");
+ rc.onCompletion().onCompleteOnly().to("mock:complete");
+ rc.onCompletion().onFailureOnly().to("mock:failure");
+ }
+ });
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start").routeConfigurationId("myconfig")
+ .to("mock:result");
+
+ from("direct:fail").routeConfigurationId("myconfig")
+ .throwException(new IllegalArgumentException("Boom"));
+ }
+ });
+ context.start();
+
+ // success case
+ getMockEndpoint("mock:complete").expectedMessageCount(1);
+ getMockEndpoint("mock:failure").expectedMessageCount(0);
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+
+ template.sendBody("direct:start", "Hello World");
+ assertMockEndpointsSatisfied();
+
+ // failure case
+ resetMocks();
+ getMockEndpoint("mock:complete").expectedMessageCount(0);
+ getMockEndpoint("mock:failure").expectedMessageCount(1);
+
+ assertThrows(Exception.class, () -> template.sendBody("direct:fail",
"Boom"));
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testLocalConfigurationBeforeConsumerFiringCount() throws
Exception {
+ // BeforeConsumer mode, both routes opt in. onCompletion should fire
only once
+ // and the consumer route should own the firing (so onCompletion sees
state set
+ // by the consumer after the direct:processor call returns).
+ AtomicInteger counter = new AtomicInteger();
+ // Capture what the onCompletion processor actually sees for
"fromConsumer";
+ // the consumer sets this header AFTER calling direct:processor, so if
the
+ // onCompletion fires early (in the sub-route) it will see null, not
"yes".
+ AtomicReference<String> seen = new AtomicReference<>();
+ context.addRoutes(new RouteConfigurationBuilder() {
+ @Override
+ public void configuration() {
+
routeConfiguration("myconfig").onCompletion().modeBeforeConsumer()
+ .process(e -> {
+ counter.incrementAndGet();
+ seen.set(e.getMessage().getHeader("fromConsumer",
String.class));
+ })
+ .setHeader("done", constant("yes"));
+ }
+ });
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ // consumer sets a header AFTER the sub-route call;
+ // since BeforeConsumer defers to the consumer route, it fires
last
+ // and sees this header on the exchange (validated via the
reply).
+ from("direct:consumer").routeConfigurationId("myconfig")
+ .to("direct:processor")
+ .setHeader("fromConsumer", constant("yes"));
+
+ from("direct:processor").routeConfigurationId("myconfig")
+ .to("mock:result");
+ }
+ });
+ context.start();
+
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+
+ Exchange result = template.request("direct:consumer",
+ e -> e.getMessage().setBody("Hello World"));
+
+ assertMockEndpointsSatisfied();
+ // onCompletion must fire exactly once (not once per opted-in route)
+ assertEquals(1, counter.get(), "onCompletion should fire exactly
once");
+ // deferral: onCompletion was deferred to the consumer route, so it
ran after
+ // the consumer set "fromConsumer" — the processor must have seen "yes"
+ assertEquals("yes", seen.get(),
+ "BeforeConsumer onCompletion should run after the consumer
route sets fromConsumer");
+ // onCompletion itself sets the 'done' header; it should be visible on
the reply
+ assertEquals("yes", result.getMessage().getHeader("done"));
+ }
+
}
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 0874e9de60df..5f179794fc64 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
@@ -170,6 +170,21 @@ onCompletion can therefore read a spooled body, also with
`parallelProcessing`.
an on completion with the order `Ordered.LOWEST` that was added after the
onCompletion EIP (for example the close
of a JPA `EntityManager` used later in the route) used to run before the
onCompletion, and now runs after it.
+=== camel-core - routeConfiguration named onCompletion fires once per exchange
in the Java DSL
+
+In the Java DSL, `routeConfiguration("myconfig").onCompletion()` previously
set `routeScoped=false`
+unconditionally, which caused the `onCompletion` to be skipped when the
opted-in route was called
+indirectly (for example via `direct:` from a REST DSL consumer route).
+
+The behavior is now corrected: a named route configuration
(`routeConfiguration("id")`) keeps
+`routeScoped=true` so that the `onCompletion` fires correctly regardless of
which route is the
+actual consumer. A per-exchange deduplication key ensures the `onCompletion`
fires exactly once
+per exchange, even when multiple routes opt in to the same named
configuration, or when the
+opted-in route is called several times in the same exchange.
+
+If your application relied on the (broken) behavior where `onCompletion` was
silently skipped for
+named configurations in indirect call chains, you may now observe it firing
where it did not before.
+
=== @PropertyInject - an invalid property value is an error (Breaking change)
When a property injected with `@PropertyInject` has a value that cannot be
converted to the type of the field or
