davsclaus commented on code in PR #27430:
URL: https://github.com/apache/camel/pull/27430#discussion_r4206866475
##########
core/camel-core-processor/src/main/java/org/apache/camel/processor/OnCompletionProcessor.java:
##########
@@ -504,6 +513,22 @@ private boolean shouldSkip(Exchange exchange, boolean
onCompleteOrOnFailureOnly)
if (routeIds == null || !routeIds.contains(routeId)) {
return true;
}
+ // named configuration: fire only once per exchange across all
opted-in routes.
+ // Multiple OnCompletionProcessor instances may share the same
configurationId
+ // (when several routes use the same routeConfigurationId);
only the first to
+ // fire records the id and the others skip.
+ if (configurationId != null) {
+ Set<String> firedConfigIds
+ =
exchange.getProperty(ExchangePropertyKey.ON_COMPLETION_FIRED_CONFIG_IDS,
Set.class);
+ if (firedConfigIds != null &&
firedConfigIds.contains(configurationId)) {
+ return true;
+ }
+ if (firedConfigIds == null) {
+ firedConfigIds = new HashSet<>();
+
exchange.setProperty(ExchangePropertyKey.ON_COMPLETION_FIRED_CONFIG_IDS,
firedConfigIds);
+ }
+ firedConfigIds.add(configurationId);
Review Comment:
The key is the configuration id, so all the onCompletions of the same
configuration share it: with `onCompleteOnly()` and `onFailureOnly()` in one
configuration, the onCompleteOnly synchronization records `cfg` here and then
skips at the `onCompleteOrOnFailureOnly` check below, and the onFailureOnly one
is skipped by this dedup, so a failed exchange runs neither. Could the key be
per `OnCompletionDefinition`, and be recorded only once the onCompletion is
really going to run (after the mode and `onWhen` checks)?
##########
core/camel-core-processor/src/main/java/org/apache/camel/processor/OnCompletionProcessor.java:
##########
@@ -587,6 +614,21 @@ public void onAfterRoute(Route route, Exchange exchange) {
return;
}
+ // named configuration: fire only once per exchange across
all opted-in routes
+ if (routeScoped && configurationId != null) {
+ @SuppressWarnings("unchecked")
+ Set<String> firedConfigIds
+ =
exchange.getProperty(ExchangePropertyKey.ON_COMPLETION_FIRED_CONFIG_IDS,
Set.class);
+ if (firedConfigIds != null &&
firedConfigIds.contains(configurationId)) {
+ return; // another processor for the same named
config already fired
+ }
+ if (firedConfigIds == null) {
+ firedConfigIds = new HashSet<>();
+
exchange.setProperty(ExchangePropertyKey.ON_COMPLETION_FIRED_CONFIG_IDS,
firedConfigIds);
+ }
+ firedConfigIds.add(configurationId);
Review Comment:
Same key issue as in `shouldSkip`. In addition, when the consumer route and
the `direct:` route both opt in, the sub-route's `onAfterRoute` runs first and
wins, so the BeforeConsumer onCompletion now runs before the rest of the
consumer route (on main it runs at the end of the consumer route). State set by
the consumer after the call, or an `onWhen` that depends on it, is then not
seen.
##########
core/camel-core-processor/src/main/java/org/apache/camel/processor/OnCompletionProcessor.java:
##########
@@ -73,14 +74,17 @@ 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();
public OnCompletionProcessor(CamelContext camelContext, Processor
processor, ExecutorService executorService,
boolean shutdownExecutorService,
boolean onCompleteOnly, boolean
onFailureOnly, Predicate onWhen, boolean useOriginalBody,
- boolean afterConsumer, boolean routeScoped) {
+ boolean afterConsumer, boolean routeScoped,
String configurationId) {
Review Comment:
Nit: this is a public constructor; maybe keep the previous signature as an
overload that delegates with `null`, so code that builds the processor directly
still compiles.
##########
core/camel-core/src/test/java/org/apache/camel/model/RoutesConfigurationOnCompletionTest.java:
##########
@@ -122,4 +125,166 @@ public void configure() {
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"));
Review Comment:
This checks that the header is there but not how many times the onCompletion
ran; a counter would pin the once-per-exchange rule for BeforeConsumer. Tests
for a `routeConfiguration("*")` configuration and for two onCompletions
(onCompleteOnly + onFailureOnly) in the same configuration would also help.
##########
core/camel-core-reifier/src/main/java/org/apache/camel/reifier/OnCompletionReifier.java:
##########
@@ -71,9 +71,17 @@ public Processor createProcessor() throws Exception {
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();
Review Comment:
`getRouteConfiguration()` is only set by the Java DSL
(`RouteConfigurationDefinition.onCompletion()`); YAML and XML go through
`setOnCompletions()`, so this is null for them and their named configurations
still fire once per visited route. Setting the key where the route
configurations are merged into the routes would cover all DSLs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
