davsclaus commented on code in PR #28260:
URL: https://github.com/apache/camel/pull/28260#discussion_r4346373046
##########
components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletComponent.java:
##########
@@ -484,13 +543,120 @@ protected void doInit() throws Exception {
super.doInit();
}
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ ScheduledExecutorService executor = routeRemover;
+ if (executor == null || executor.isShutdown()) {
+ // the thread is created on first use
+ routeRemover =
getCamelContext().getExecutorServiceManager().newSingleThreadScheduledExecutor(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:
stopping a route holds the lock of the
+ * CamelContext while the route shuts down gracefully. 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, which
+ * interrupts that thread. So the removal waits:
+ * <ul>
+ * <li>until the exchange whose call is completing on this thread is done,
when the endpoint is shut down inside the
+ * producer callback (toD);</li>
+ * <li>and until no thread of the thread pools that the route's processors
created (threads(), and the other EIPs
+ * with an executor service) runs a task. This covers an endpoint that is
shut down where the exchange of the
+ * calling route is not known: recipientList with cacheSize=-1 (the
endpoint is shut down when the producer is
+ * released, after the call), routingSlip with cacheSize=-1 (its copy of
the exchange has no unit of work) and
+ * pollEnrich. The route is used only by the endpoint that created it, so
its thread pools stay idle once the
+ * calling route has left them.</li>
+ * </ul>
+ * Not covered: a thread pool that a component of the route creates for
itself (such as the worker pool of an
+ * asynchronous Kafka producer, which waits for its running tasks when it
stops, up to its shutdown timeout), and an
+ * executor service that is not a {@link ThreadPoolExecutor} (virtual
threads), which the removal cannot tell idle.
+ *
+ * @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, 0);
+ }
+ });
+ } else {
+ submitRemoveRoute(route, 0);
+ }
+ }
+
+ private void submitRemoveRoute(Route route, long delayMillis) {
+ ScheduledExecutorService executor = routeRemover;
+ if (executor == null) {
+ // the context removes all the routes when it stops
+ return;
+ }
+ try {
+ executor.schedule(() -> doRemoveRoute(route), delayMillis,
TimeUnit.MILLISECONDS);
+ } catch (RejectedExecutionException e) {
+ // the component is shut down, and the context removes all the
routes when it stops
+ LOG.debug("Cannot remove route with id={} as the component is shut
down", route.getRouteId());
+ }
+ }
+
+ private void doRemoveRoute(Route route) {
+ CamelContext context = getCamelContext();
+ String routeId = route.getRouteId();
+ if (!context.isStarted() || context.isStopping() ||
context.getRoute(routeId) != route) {
+ // the context removes all the routes when it stops, and a route
that is gone or was replaced is not ours
+ return;
+ }
+ if (isThreadPoolActive(routeId)) {
+ // the calling route may still run on a thread of the route's
thread pools: stopping the route now would
+ // shut the pool down under it
+ LOG.trace("Route with id={} still has an active thread: removing
it later", routeId);
+ submitRemoveRoute(route, ROUTE_REMOVAL_RETRY_MILLIS);
Review Comment:
I tried this with a probe test on ee985a1c22, and it needs a correction:
steady input alone is not enough. The first removal attempt runs at once, so
with fast steps the pool is usually idle at that moment and the route is
removed. It happens when the pool is busy at the moment the consumer route is
removed (a slow step: an HTTP call, a database, here a `delay`). After that,
every new message waits up to 30 s on `kamelet:sink`, so the pool does not go
idle again.
| Source template (`threads(2)`), consumer route removed at runtime | Result
|
|---|---|
| `timer:idle?period=60000` | Kamelet route removed within about 1 s |
| `timer:busy?period=100` | Kamelet route removed within about 1 s |
| `timer:slow?period=100` + `delay(500).syncDelayed()` | still there after
45 s (3 out of 3 runs). With TRACE on `KameletComponent`: 434 times "Route with
id=slow-ticker-1 still has an active thread: removing it later" in 45 s |
The reproducer, in case it helps:
```java
class KameletSourceRemovalProbeTest extends CamelTestSupport {
@Override
protected CamelContext createCamelContext() throws Exception {
CamelContext context = super.createCamelContext();
context.getShutdownStrategy().setTimeout(2);
return context;
}
@Test
void busySlowSourceWithThreads() throws Exception {
context.addRoutes(new RouteBuilder() {
@Override
public void configure() {
from("kamelet:slow-ticker").routeId("consumer-slow").to("mock:consumer-slow");
}
});
MockEndpoint mock = getMockEndpoint("mock:consumer-slow");
mock.expectedMinimumMessageCount(1);
mock.assertIsSatisfied();
context.getRouteController().stopRoute("consumer-slow");
assertThat(context.removeRoute("consumer-slow")).isTrue();
await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> assertThat(
context.getRoutes().stream().filter(Route::isCreatedByKamelet).count()).isZero());
}
@Override
protected RoutesBuilder createRouteBuilder() {
return new RouteBuilder() {
@Override
public void configure() {
routeTemplate("slow-ticker")
.from("timer:slow?period=100")
.threads(2)
.delay(500).syncDelayed()
.setBody().constant("tick")
.to("kamelet:sink");
}
};
}
}
```
The route was never removed before this PR either, so this is not a
regression and stays optional. Skipping the idle wait when the endpoint was the
input of a removed route would cover it, since no calling route runs on that
pool. Fine as a follow-up JIRA too.
_Claude Code on behalf of @davsclaus. This comment was generated by an AI
agent and may contain inaccuracies. Please verify before applying._
--
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]
