davsclaus commented on code in PR #28260:
URL: https://github.com/apache/camel/pull/28260#discussion_r4346134009
##########
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:
Optional: this retries every 100 ms with no limit. For the endpoint of a
removed route that consumed from a source Kamelet, the Kamelet's route still
runs its own consumer. While the consumer route is gone, its messages wait on
`kamelet:sink` (`block=true`, 30 s timeout by default), so a pool from
`threads()` in that template may never be idle under steady input. The route is
then never removed, and this task runs until the context stops. Maybe stop the
route anyway after the shutdown strategy timeout, or give up with a DEBUG log
after a number of tries.
--
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]
