allthingssecurity commented on code in PR #28260: URL: https://github.com/apache/camel/pull/28260#discussion_r4345716380
########## components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletToDRoutesNotKeptTest.java: ########## @@ -0,0 +1,282 @@ +/* + * 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.component.kamelet; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import org.apache.camel.Exchange; +import org.apache.camel.Route; +import org.apache.camel.RoutesBuilder; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.test.junit6.CamelTestSupport; +import org.awaitility.core.ConditionTimeoutException; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +/** + * A Kamelet called with toD and cacheSize=-1 (the documented setting for a dynamic endpoint that is different for every + * message) gets a new endpoint for every call, which creates a route from the template. The endpoint is discarded after + * the call, so its route must be removed as well, otherwise every call keeps one more route. + */ +class KameletToDRoutesNotKeptTest extends CamelTestSupport { + + private static final int MESSAGES = 50; + + @Test + void routesOfDiscardedKameletEndpointsAreRemoved() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:result"); + mock.expectedMessageCount(MESSAGES); + int endpoints = context.getEndpointRegistry().size(); + + for (int i = 0; i < MESSAGES; i++) { + template.sendBodyAndHeader("direct:prototype", "body", "id", "id" + i); + } + mock.assertIsSatisfied(); + for (int i = 0; i < MESSAGES; i++) { + assertThat(mock.getReceivedExchanges().get(i).getMessage().getBody()).isEqualTo("id" + i + "-body"); + } + + // the routes are removed on a thread of the component, once the exchange is done + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertThat(kameletRoutes()).isZero()); + // and their endpoints (kamelet:source) with them + assertThat(context.getEndpointRegistry().values()).hasSize(endpoints); Review Comment: Fixed in ee985a1c2229. The endpoint-registry assertion is now in the same `untilAsserted` block as the route count, with a comment saying why: `removeRoute` takes the route out of `getRoutes()` before `doRemove` removes its private endpoints. _Claude Code on behalf of allthingssecurity_ ########## components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletComponent.java: ########## @@ -484,13 +536,86 @@ protected void doInit() throws Exception { super.doInit(); } + @Override + protected void doStart() throws Exception { + super.doStart(); + ExecutorService executor = routeRemover; + if (executor == null || executor.isShutdown()) { + // the threads are created on first use + routeRemover = getCamelContext().getExecutorServiceManager().newSingleThreadExecutor(this, "KameletRouteRemover"); + } + } + + /** + * Removes the route a Kamelet endpoint created from its template while the context runs, when that endpoint is shut + * down (it is not shut down while the context runs unless it was used once and discarded, or its route consumed + * from the Kamelet and was removed). + * <p/> + * The route is stopped and removed on a thread of this component, after the exchange whose call is completing is + * done: the calling route can continue on a thread that the Kamelet's route owns (threads(), an asynchronous + * producer), and stopping the route shuts that thread pool down. Stopping a route also holds the lock of the + * CamelContext while the route shuts down gracefully. + * + * @param route the route the endpoint created + * @param exchange the exchange whose call of the endpoint is completing on this thread, or {@code null} + */ + private void removeCreatedRoute(Route route, Exchange exchange) { + if (exchange != null && exchange.getUnitOfWork() != null) { + exchange.getExchangeExtension().addOnCompletion(new SynchronizationAdapter() { + @Override + public void onDone(Exchange done) { + submitRemoveRoute(route); + } + }); + } else { + submitRemoveRoute(route); Review Comment: Fixed in ee985a1c2229, with the test you suggested. It did matter: `kameletRouteIsKeptWhileTheCallingRouteRunsOnItsThreadAfterARecipientList` (recipientList `cacheSize=-1`, the `pooled` template) fails without the fix. So does the same test with routingSlip `cacheSize=-1`: its copy has no unit of work. In both, the Kamelet's route was removed while the calling route ran on its `threads()` thread. Where the endpoint is shut down, the calling exchange is not known (the recipient list copy carries only its id as `CamelCorrelationId`). So the removal now also waits until the route's own thread pools are idle. The `LifecycleHandler` tracks the pools that the route's EIPs create (`onThreadPoolAdd` with the route id, `onThreadPoolRemove`). The remover is now a single-thread scheduled executor. While one of those pools has an active task, it checks again after 100 ms before it stops the route. The route is used only by the endpoint that created it, so its pools stay idle once the calling route has left the thread. This applies to every path, toD included, on top of the on-completion wait. What it does not cover, as the javadoc of `removeCreatedRoute` says: a pool that a component of the route creates for itself, such as the worker pool of an async kafka producer (no route id; `KafkaProducer.doStop` waits for running tasks up to its shutdown timeout), and virtual-thread executors (they are not a `ThreadPoolExecutor`, so there is no active count). There is a follow-up JIRA draft for those two. _Claude Code on behalf of allthingssecurity_ -- 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]
