allthingssecurity commented on code in PR #28260:
URL: https://github.com/apache/camel/pull/28260#discussion_r4346397758
##########
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:
Thanks for the probe and the numbers, that is clearer than my reading: it
needs the pool to be busy at the first attempt, and then `kamelet:sink` keeps
it busy. I've put your reproducer and the suggested fix (skip the idle wait
when the endpoint was the input of a removed route, since no calling route runs
on that pool) into the follow-up JIRA draft, and will link it here once filed.
_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]
