This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new a910bc51d95b CAMEL-26645: camel-core - Supervising route controller:
route consumers get the application class loader as TCCL (#29263)
a910bc51d95b is described below
commit a910bc51d95bb06d1e99c702931ea785e9408eaf
Author: Serdar Gökay <[email protected]>
AuthorDate: Thu Nov 5 22:50:06 2026 +0300
CAMEL-26645: camel-core - Supervising route controller: route consumers get
the application class loader as TCCL (#29263)
A consumer creates its threads when it is started, and they inherit the
TCCL of the thread starting the route.
AbstractCamelContext sets the application class loader as TCCL in
doLifecycleChange(), which covers doStart() only. The
supervising route controller starts its routes from
onCamelContextFullyStarted, after that scope has closed, and
camel-kafka creates its consumer threads there. Under camel run, lookups
through the TCCL on those threads miss the
downloaded libraries, and camel-mail silently loses attachments.
Use doLifecycleChange() also around
InternalRouteStartupManager.safelyStartRouteServices, which every route start
passes
through, and around the onCamelContextFullyStarted loop.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/impl/engine/AbstractCamelContext.java | 11 +-
.../impl/engine/InternalRouteStartupManager.java | 4 +-
.../RouteStartApplicationClassLoaderTest.java | 197 +++++++++++++++++++++
3 files changed, 206 insertions(+), 6 deletions(-)
diff --git
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java
index 4e0673a16ca1..09dee6760730 100644
---
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java
+++
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java
@@ -2530,13 +2530,14 @@ public abstract class AbstractCamelContext extends
BaseService
// has started (here at the end)
EventHelper.notifyCamelContextStarted(this);
- // now call the startup listeners where the routes has been started
- for (StartupListener startup : startupListeners) {
- try {
+ // now call the startup listeners where the routes has been started,
with the TCCL as during the lifecycle
+ // callbacks, as they may start routes or create the threads of
consumers (supervising route controller, kafka)
+ try (AutoCloseable ignored = doLifecycleChange()) {
+ for (StartupListener startup : startupListeners) {
startup.onCamelContextFullyStarted(this, isStarted());
- } catch (Exception e) {
- throw RuntimeCamelException.wrapRuntimeException(e);
}
+ } catch (Exception e) {
+ throw RuntimeCamelException.wrapRuntimeException(e);
}
}
diff --git
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/InternalRouteStartupManager.java
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/InternalRouteStartupManager.java
index 8eafc028ffc8..41e74b81a059 100644
---
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/InternalRouteStartupManager.java
+++
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/InternalRouteStartupManager.java
@@ -172,7 +172,9 @@ final class InternalRouteStartupManager {
Collection<RouteService> routeServices)
throws Exception {
lock.lock();
- try {
+ // the TCCL is only set during the lifecycle callbacks of
CamelContext, routes also start after them (supervising
+ // route controller, manual start), and the threads a consumer creates
inherit it
+ try (AutoCloseable ignored = camelContext.doLifecycleChange()) {
// list of inputs to start when all the routes have been prepared
for
// starting
// we use a tree map so the routes will be ordered according to
startup
diff --git
a/core/camel-core/src/test/java/org/apache/camel/impl/engine/RouteStartApplicationClassLoaderTest.java
b/core/camel-core/src/test/java/org/apache/camel/impl/engine/RouteStartApplicationClassLoaderTest.java
new file mode 100644
index 000000000000..e627aa4acfbb
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/impl/engine/RouteStartApplicationClassLoaderTest.java
@@ -0,0 +1,197 @@
+/*
+ * 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.impl.engine;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Consumer;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Processor;
+import org.apache.camel.StartupListener;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.component.timer.TimerConsumer;
+import org.apache.camel.component.timer.TimerEndpoint;
+import org.apache.camel.spi.SupervisingRouteController;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+/**
+ * A consumer creates its threads when it is started, and they inherit the
thread context class loader of the thread
+ * starting the route. Libraries use it to look up their services, so the
application class loader must be in place
+ * however a route is started, not only during the lifecycle callbacks of
CamelContext.
+ */
+public class RouteStartApplicationClassLoaderTest extends ContextTestSupport {
+
+ private static final String TIMER_URI = "timer:tccl?delay=1&repeatCount=1";
+
+ private final ClassLoader applicationClassLoader = new
ClassLoader(getClass().getClassLoader()) {
+ };
+ private final AtomicReference<ClassLoader> consumerThreadClassLoader = new
AtomicReference<>();
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ @Test
+ public void routeStartedByCamelContext() throws Exception {
+ context.setApplicationContextClassLoader(applicationClassLoader);
+ context.addRoutes(routeFrom(TIMER_URI, true));
+
+ startAndAssertConsumerThreadClassLoader(applicationClassLoader);
+ }
+
+ @Test
+ public void supervisedRoute() throws Exception {
+ context.setApplicationContextClassLoader(applicationClassLoader);
+ context.getRouteController().supervising();
+ context.addRoutes(routeFrom(TIMER_URI, true));
+
+ startAndAssertConsumerThreadClassLoader(applicationClassLoader);
+ }
+
+ @Test
+ public void supervisedRouteWithInitialDelay() throws Exception {
+ context.setApplicationContextClassLoader(applicationClassLoader);
+ context.getRouteController().supervising().setInitialDelay(10);
+ context.addRoutes(routeFrom(TIMER_URI, true));
+
+ startAndAssertConsumerThreadClassLoader(applicationClassLoader);
+ }
+
+ @Test
+ public void routeExcludedFromSupervision() throws Exception {
+ context.setApplicationContextClassLoader(applicationClassLoader);
+ context.getRouteController().supervising().setExcludeRoutes("tccl");
+ context.addRoutes(routeFrom(TIMER_URI, true));
+
+ startAndAssertConsumerThreadClassLoader(applicationClassLoader);
+ }
+
+ @Test
+ public void supervisedRouteRestartedAfterFailedStart() throws Exception {
+ context.setApplicationContextClassLoader(applicationClassLoader);
+ SupervisingRouteController src =
context.getRouteController().supervising();
+ src.setBackOffDelay(10);
+ src.setBackOffMaxAttempts(3);
+ context.addRoutes(routeFrom(failsOnFirstStart(), true));
+
+ startAndAssertConsumerThreadClassLoader(applicationClassLoader);
+ }
+
+ @Test
+ public void routeStartedManuallyAfterCamelContextStart() throws Exception {
+ context.setApplicationContextClassLoader(applicationClassLoader);
+ context.addRoutes(routeFrom(TIMER_URI, false));
+ context.start();
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+ ClassLoader caller = Thread.currentThread().getContextClassLoader();
+
+ context.getRouteController().startRoute("tccl");
+
+ MockEndpoint.assertIsSatisfied(context);
+ assertSame(applicationClassLoader, consumerThreadClassLoader.get());
+ assertSame(caller, Thread.currentThread().getContextClassLoader());
+ }
+
+ @Test
+ public void
routeWithoutApplicationClassLoaderKeepsTheClassLoaderOfTheCaller() throws
Exception {
+ context.getRouteController().supervising();
+ context.addRoutes(routeFrom(TIMER_URI, true));
+
+
startAndAssertConsumerThreadClassLoader(Thread.currentThread().getContextClassLoader());
+ }
+
+ @Test
+ public void startupListenerWhenCamelContextIsFullyStarted() throws
Exception {
+ context.setApplicationContextClassLoader(applicationClassLoader);
+ AtomicReference<ClassLoader> listenerClassLoader = new
AtomicReference<>();
+ context.addStartupListener(new StartupListener() {
+ @Override
+ public void onCamelContextStarted(CamelContext camelContext,
boolean alreadyStarted) {
+ // noop
+ }
+
+ @Override
+ public void onCamelContextFullyStarted(CamelContext camelContext,
boolean alreadyStarted) {
+
listenerClassLoader.set(Thread.currentThread().getContextClassLoader());
+ }
+ });
+ ClassLoader caller = Thread.currentThread().getContextClassLoader();
+
+ context.start();
+
+ assertSame(applicationClassLoader, listenerClassLoader.get());
+ assertSame(caller, Thread.currentThread().getContextClassLoader());
+ }
+
+ private void startAndAssertConsumerThreadClassLoader(ClassLoader expected)
throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+ ClassLoader caller = Thread.currentThread().getContextClassLoader();
+
+ context.start();
+
+ MockEndpoint.assertIsSatisfied(context);
+ assertSame(expected, consumerThreadClassLoader.get());
+ assertSame(caller, Thread.currentThread().getContextClassLoader());
+ }
+
+ private RouteBuilder routeFrom(String uri, boolean autoStartup) {
+ return routeFrom(context.getEndpoint(uri), autoStartup);
+ }
+
+ private RouteBuilder routeFrom(Endpoint endpoint, boolean autoStartup) {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from(endpoint).routeId("tccl").autoStartup(autoStartup)
+ .process(e ->
consumerThreadClassLoader.set(Thread.currentThread().getContextClassLoader()))
+ .to("mock:result");
+ }
+ };
+ }
+
+ private Endpoint failsOnFirstStart() {
+ AtomicBoolean failed = new AtomicBoolean();
+ TimerEndpoint endpoint = new TimerEndpoint("timer:tccl",
context.getComponent("timer"), "tccl") {
+ @Override
+ public Consumer createConsumer(Processor processor) throws
Exception {
+ Consumer consumer = new TimerConsumer(this, processor) {
+ @Override
+ protected void doStart() throws Exception {
+ if (failed.compareAndSet(false, true)) {
+ throw new IllegalStateException("Cannot start");
+ }
+ super.doStart();
+ }
+ };
+ configureConsumer(consumer);
+ return consumer;
+ }
+ };
+ endpoint.setDelay(1);
+ endpoint.setRepeatCount(1);
+ return endpoint;
+ }
+}
