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 8b1cb4437b12 CAMEL-26310: camel-core - Do not use a component created
on first use before it has started (#28710)
8b1cb4437b12 is described below
commit 8b1cb4437b123542418a24be7722319510494f94
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 31 01:50:35 2026 +0530
CAMEL-26310: camel-core - Do not use a component created on first use
before it has started (#28710)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/impl/engine/AbstractCamelContext.java | 76 ++++-
.../camel/impl/ComponentStartedBeforeUseTest.java | 374 +++++++++++++++++++++
2 files changed, 448 insertions(+), 2 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 39c3206fe97d..4e0673a16ca1 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
@@ -240,6 +240,13 @@ public abstract class AbstractCamelContext extends
BaseService
private final Set<AutoMockInterceptStrategy> autoMockInterceptStrategies =
ConcurrentHashMap.newKeySet();
private final GlobalEndpointConfiguration globalEndpointConfiguration =
new DefaultGlobalEndpointConfiguration();
private final Map<String, Component> components = new
ConcurrentHashMap<>();
+ // the components that getComponent created with autoStart=false: their
caller configures and then starts them
+ // (such as RestProducerFactoryHelper.setupComponent), so another thread
must not start them meanwhile
+ private final Map<String, Component> componentsCreatedWithoutAutoStart =
new ConcurrentHashMap<>();
+ // the locks under which getComponent starts a component created on first
use, so that it is started only once
+ private final Map<String, Lock> componentStartLocks = new
ConcurrentHashMap<>();
+ // whether doStartCamel has started the components it has (until then it
starts them itself, as startup steps)
+ private volatile boolean componentsStarted;
private final Set<Route> routes = new LinkedHashSet<>();
private final List<StartupListener> startupListeners = new
CopyOnWriteArrayList<>();
private final Map<String, Language> languages = new ConcurrentHashMap<>();
@@ -559,11 +566,20 @@ public abstract class AbstractCamelContext extends
BaseService
try {
// Flag used to mark a component of being created.
final AtomicBoolean created = new AtomicBoolean();
+ // whether its creator starts the component also when it is not
new (such as a stopped bean from the registry)
+ final AtomicBoolean startAlways = new AtomicBoolean();
// atomic operation to get/create a component. Avoid global locks.
final Component component = components.computeIfAbsent(name, comp
-> {
created.set(true);
- return initComponent(name, autoCreateComponents);
+ Component answer = initComponent(name, autoCreateComponents);
+ startAlways.set(answer != null && !isNotStartedYet(answer));
+ if (answer != null && !autoStart) {
+ // marked before other threads can get it
+ // (kept after the creator starts it: not consulted again,
removed by removeComponent or context stop)
+ componentsCreatedWithoutAutoStart.put(name, answer);
+ }
+ return answer;
});
// Start the component after its creation as if it is a component
proxy
@@ -573,8 +589,19 @@ public abstract class AbstractCamelContext extends
BaseService
// lets start it up.
final StartupStepRecorder startupStepRecorder =
camelContextExtension.getStartupStepRecorder();
StartupStep step =
startupStepRecorder.beginStep(Component.class, name, "Start Component");
- startService(component);
+ startComponentCreatedOnFirstUse(name, component,
startAlways.get());
startupStepRecorder.endStep(step);
+ } else if (autoStart && (isStarted() || (isStarting() &&
componentsStarted)) && isNotStartedYet(component)
+ && componentsCreatedWithoutAutoStart.get(name) !=
component) {
+ // another thread has just created the component and may still
be starting it: start it (or wait for
+ // the other thread to have started it), so the component is
not used before it has been started.
+ // A component created with autoStart=false is left to its
creator, which configures it first, and
+ // until doStartCamel has started its components, it starts
them itself.
+ // This thread waits for the start lock of the component, also
when it holds the lock of this
+ // CamelContext (such as when a route is started): the doStart
of a component must therefore not take
+ // the lock of this CamelContext (such as by adding or
starting routes), nor get a component whose
+ // doStart gets it back, or the threads deadlock.
+ startComponentCreatedOnFirstUse(name, component, false);
}
return component;
@@ -586,6 +613,45 @@ public abstract class AbstractCamelContext extends
BaseService
}
}
+ /**
+ * Starts a component that getComponent has created on first use. Its
creator, and the threads that get it
+ * meanwhile, start it under the same lock, so it is started once: a
thread that gets the lock after the component
+ * has started, or has failed to start, gets the component as it is (only
the thread that started it gets the
+ * failure). The creator of a component that was not new when it was
created (a component that is not a BaseService,
+ * or a bean from the registry that has been stopped) starts it always, as
before.
+ */
+ private void startComponentCreatedOnFirstUse(String name, Component
component, boolean always) throws Exception {
+ Lock lock = componentStartLocks.computeIfAbsent(name, k -> new
ReentrantLock());
+ lock.lock();
+ try {
+ if (always || isNotStartedYet(component)) {
+ startService(component);
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Whether the component has not been started yet, or is being started. A
component that has been stopped, or has
+ * failed to start, is not.
+ */
+ private static boolean isNotStartedYet(Component component) {
+ if (component instanceof BaseService bs) {
+ if (bs.isStarted()) {
+ // the common case
+ return false;
+ }
+ if (bs.isNew() || bs.isBuild()) {
+ return true;
+ }
+ ServiceStatus status = bs.getStatus();
+ return status == ServiceStatus.Initializing || status ==
ServiceStatus.Initialized
+ || status == ServiceStatus.Starting;
+ }
+ return false;
+ }
+
/**
* Function to initialize a component and auto start. Returns null if the
autoCreateComponents is disabled
*/
@@ -661,6 +727,8 @@ public abstract class AbstractCamelContext extends
BaseService
public Component removeComponent(String componentName) {
Component oldComponent = components.remove(componentName);
if (oldComponent != null) {
+ componentsCreatedWithoutAutoStart.remove(componentName,
oldComponent);
+ componentStartLocks.remove(componentName);
try {
stopServices(oldComponent);
} catch (Exception e) {
@@ -3201,6 +3269,7 @@ public abstract class AbstractCamelContext extends
BaseService
startupStepRecorder.endStep(step);
}
}
+ componentsStarted = true;
if (!startupListeners.isEmpty()) {
StartupStep subStep
@@ -3421,6 +3490,9 @@ public abstract class AbstractCamelContext extends
BaseService
InternalServiceManager.shutdownServices(this, components.values());
components.clear();
+ componentsCreatedWithoutAutoStart.clear();
+ componentStartLocks.clear();
+ componentsStarted = false;
InternalServiceManager.shutdownServices(this, languages.values());
languages.clear();
diff --git
a/core/camel-core/src/test/java/org/apache/camel/impl/ComponentStartedBeforeUseTest.java
b/core/camel-core/src/test/java/org/apache/camel/impl/ComponentStartedBeforeUseTest.java
new file mode 100644
index 000000000000..eca34d1cebdf
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/impl/ComponentStartedBeforeUseTest.java
@@ -0,0 +1,374 @@
+/*
+ * 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;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Component;
+import org.apache.camel.Consumer;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.StartupListener;
+import org.apache.camel.StartupStep;
+import org.apache.camel.support.DefaultComponent;
+import org.apache.camel.support.DefaultEndpoint;
+import org.apache.camel.support.service.ServiceHelper;
+import org.apache.camel.support.startup.DefaultStartupStepRecorder;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A component that is created on first use, after CamelContext has started or
while it starts, is started by the thread
+ * that created it. Another thread that gets the same component meanwhile must
not create endpoints from it before it
+ * has started.
+ */
+class ComponentStartedBeforeUseTest extends ContextTestSupport {
+
+ private final CountDownLatch starting = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+ private final CountDownLatch created = new CountDownLatch(2);
+ private final CountDownLatch resumeCreator = new CountDownLatch(1);
+ private final SlowComponent component = new SlowComponent();
+
+ @Override
+ @AfterEach
+ public void tearDown() throws Exception {
+ release.countDown();
+ resumeCreator.countDown();
+ super.tearDown();
+ }
+
+ @Test
+ void testOtherThreadWaitsForTheComponentToStart() throws Exception {
+ context.getRegistry().bind("slow", component);
+ resolveFromTwoThreads(context);
+ }
+
+ @Test
+ void testOtherThreadWaitsForTheComponentToStartWhileCamelContextStarts()
throws Exception {
+ // the consumers of the routes that have started can send while
CamelContext is still starting
+ CamelContext startingContext = new DefaultCamelContext();
+ startingContext.getRegistry().bind("slow", component);
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+ startingContext.addStartupListener(new StartupListener() {
+ @Override
+ public void onCamelContextStarting(CamelContext camelContext,
boolean alreadyStarted) {
+ try {
+ assertTrue(camelContext.isStarting());
+ resolveFromTwoThreads(camelContext);
+ } catch (Throwable e) {
+ failure.set(e);
+ }
+ }
+
+ @Override
+ public void onCamelContextStarted(CamelContext camelContext,
boolean alreadyStarted) {
+ // noop
+ }
+ });
+ try {
+ startingContext.start();
+ assertNull(failure.get(), () -> String.valueOf(failure.get()));
+ } finally {
+ startingContext.stop();
+ }
+ }
+
+ @Test
+ void testOtherThreadStartsTheComponentThatItsCreatorHasNotStartedYet()
throws Exception {
+ // the creator thread has created the component (BUILT), and has not
started it yet
+ CountDownLatch creatorPaused = new CountDownLatch(1);
+ context.getCamelContextExtension().setStartupStepRecorder(new
DefaultStartupStepRecorder() {
+ @Override
+ public StartupStep beginStep(Class<?> type, String id, String
description) {
+ if ("creator".equals(Thread.currentThread().getName()) &&
"Start Component".equals(description)) {
+ creatorPaused.countDown();
+ try {
+ resumeCreator.await(5, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ return super.beginStep(type, id, description);
+ }
+ });
+ context.getRegistry().bind("slow", component);
+ AtomicReference<Throwable> first = new AtomicReference<>();
+ Thread creator = new Thread(() -> getEndpoint(context, "slow:a",
first), "creator");
+ creator.start();
+ assertTrue(creatorPaused.await(5, TimeUnit.SECONDS));
+ assertTrue(component.isBuild(), "created, and not started yet");
+
+ AtomicReference<Throwable> second = new AtomicReference<>();
+ Thread other = new Thread(() -> getEndpoint(context, "slow:b",
second), "other");
+ other.start();
+ assertTrue(starting.await(5, TimeUnit.SECONDS), "the other thread
starts the component");
+ release.countDown();
+ resumeCreator.countDown();
+ creator.join(5000);
+ other.join(5000);
+ assertFalse(creator.isAlive());
+ assertFalse(other.isAlive());
+
+ assertNull(first.get(), () -> String.valueOf(first.get()));
+ assertNull(second.get(), () -> String.valueOf(second.get()));
+ assertEquals("other", component.startedBy);
+ assertEquals(ServiceStatus.Started, component.statusAtCreate.get("a"));
+ assertEquals(ServiceStatus.Started, component.statusAtCreate.get("b"));
+ }
+
+ @Test
+ void testComponentCreatedWithoutAutoStartIsNotStartedByAnotherThread()
throws Exception {
+ release.countDown();
+ context.getRegistry().bind("slow", component);
+ // created to be configured before it starts, as
RestProducerFactoryHelper.setupComponent does
+ assertSame(component, context.getComponent("slow", true, false));
+ assertTrue(component.isBuild());
+
+ // another thread gets the component before its creator has configured
and started it
+ AtomicReference<Component> seen = new AtomicReference<>();
+ Thread other = new Thread(() ->
seen.set(context.getComponent("slow")), "other");
+ other.start();
+ other.join(5000);
+ assertFalse(other.isAlive());
+ assertSame(component, seen.get());
+ assertFalse(component.isStarted(), "left to its creator, which
configures it first");
+
+ // the creator configures and then starts it
+ component.setAutowiredEnabled(false);
+ ServiceHelper.startService(component);
+ assertEquals(Thread.currentThread().getName(), component.startedBy);
+ }
+
+ @Test
+ void testStoppedComponentIsNotStartedAgain() {
+ // a control (as on main): only a component that has not started yet
is started, not one that has been stopped
+ // (such as by removeComponent while another thread still uses it)
+ release.countDown();
+ context.getRegistry().bind("slow", component);
+ assertNotNull(context.getEndpoint("slow:a"));
+ assertTrue(component.isStarted());
+
+ component.stop();
+ assertSame(component, context.getComponent("slow"));
+ assertTrue(component.isStopped());
+ }
+
+ @Test
+ void testOtherThreadDoesNotStartTheComponentAgainWhenItFailedToStart()
throws Exception {
+ FailingComponent failing = new FailingComponent();
+ context.getRegistry().bind("failing", failing);
+ AtomicReference<Throwable> first = new AtomicReference<>();
+ Thread creator = new Thread(() -> getEndpoint(context, "failing:a",
first), "creator");
+ creator.start();
+ // the creator thread has created the component and is starting it
+ assertTrue(starting.await(5, TimeUnit.SECONDS));
+
+ AtomicReference<Component> seen = new AtomicReference<>();
+ AtomicReference<Throwable> second = new AtomicReference<>();
+ Thread other = new Thread(() -> {
+ try {
+ seen.set(context.getComponent("failing"));
+ } catch (Throwable e) {
+ second.set(e);
+ }
+ }, "other");
+ other.start();
+ // the other thread waits for the creator to have started the component
+ await().atMost(5, TimeUnit.SECONDS).until(() -> isWaiting(other));
+
+ release.countDown();
+ creator.join(5000);
+ other.join(5000);
+ assertFalse(creator.isAlive());
+ assertFalse(other.isAlive());
+
+ assertNotNull(first.get(), "the creator gets the failure of the
start");
+ assertNull(second.get(), () -> String.valueOf(second.get()));
+ assertSame(failing, seen.get());
+ assertEquals(1, failing.starts.get(), "started once");
+ assertTrue(failing.failed());
+ }
+
+ @Test
+ void testStoppedComponentFromTheRegistryIsStartedByItsCreator() {
+ // a control (as on main): a bean from the registry that is created
again after removeComponent stopped it, is
+ // started again by the thread that creates it
+ release.countDown();
+ context.getRegistry().bind("slow", component);
+ assertNotNull(context.getEndpoint("slow:a"));
+ assertNotNull(context.removeComponent("slow"));
+ assertTrue(component.isStopped());
+
+ assertNotNull(context.getEndpoint("slow:b"));
+ assertTrue(component.isStarted());
+ assertEquals(ServiceStatus.Started, component.statusAtCreate.get("b"));
+ }
+
+ @Test
+ void testComponentAddedBeforeCamelContextStartsIsStartedByCamelContext()
throws Exception {
+ // another thread that gets the component while CamelContext starts,
before CamelContext has started its
+ // components, leaves it to CamelContext, which starts it as a startup
step
+ release.countDown();
+ CamelContext startingContext = new DefaultCamelContext();
+ startingContext.addComponent("slow", component);
+ AtomicReference<String> startedByWhenGot = new AtomicReference<>("not
got");
+ startingContext.getCamelContextExtension().setStartupStepRecorder(new
DefaultStartupStepRecorder() {
+ @Override
+ public StartupStep beginStep(Class<?> type, String id, String
description) {
+ if ("slow".equals(id) && "Start
Component".equals(description)) {
+ Thread other = new Thread(() -> {
+ startingContext.getComponent("slow");
+ startedByWhenGot.set(component.startedBy);
+ }, "other");
+ other.start();
+ try {
+ other.join(5000);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ return super.beginStep(type, id, description);
+ }
+ });
+ try {
+ startingContext.start();
+ assertNull(startedByWhenGot.get(), "not started by the other
thread");
+ assertEquals(Thread.currentThread().getName(),
component.startedBy);
+ } finally {
+ startingContext.stop();
+ }
+ }
+
+ private void resolveFromTwoThreads(CamelContext camelContext) throws
Exception {
+ AtomicReference<Throwable> first = new AtomicReference<>();
+ Thread creator = new Thread(() -> getEndpoint(camelContext, "slow:a",
first), "creator");
+ creator.start();
+ // the creator thread has created the component and is starting it
+ assertTrue(starting.await(5, TimeUnit.SECONDS));
+
+ AtomicReference<Throwable> second = new AtomicReference<>();
+ Thread other = new Thread(() -> getEndpoint(camelContext, "slow:b",
second), "other");
+ other.start();
+ // the other thread either created its endpoint already, or waits for
the component to be started
+ await().atMost(5, TimeUnit.SECONDS).until(() -> created.getCount() ==
1 || isWaiting(other));
+
+ release.countDown();
+ creator.join(5000);
+ other.join(5000);
+ assertFalse(creator.isAlive());
+ assertFalse(other.isAlive());
+
+ assertNull(first.get(), () -> String.valueOf(first.get()));
+ assertNull(second.get(), () -> String.valueOf(second.get()));
+ assertEquals(ServiceStatus.Started, component.statusAtCreate.get("a"));
+ assertEquals(ServiceStatus.Started, component.statusAtCreate.get("b"));
+ }
+
+ private static boolean isWaiting(Thread thread) {
+ Thread.State state = thread.getState();
+ return state == Thread.State.WAITING || state ==
Thread.State.TIMED_WAITING || state == Thread.State.BLOCKED;
+ }
+
+ private static void getEndpoint(CamelContext camelContext, String uri,
AtomicReference<Throwable> failure) {
+ try {
+ assertNotNull(camelContext.getEndpoint(uri));
+ } catch (Throwable e) {
+ failure.set(e);
+ }
+ }
+
+ /**
+ * A component that fails to start (such as when it cannot connect).
+ */
+ private final class FailingComponent extends DefaultComponent {
+ private final AtomicInteger starts = new AtomicInteger();
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ starts.incrementAndGet();
+ starting.countDown();
+ assertTrue(release.await(5, TimeUnit.SECONDS));
+ throw new IllegalStateException("Cannot connect");
+ }
+
+ boolean failed() {
+ return status == FAILED;
+ }
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining,
Map<String, Object> parameters) {
+ throw new UnsupportedOperationException(uri);
+ }
+ }
+
+ /**
+ * A component that creates the client its endpoints use when it starts
(as for example vertx-http creates its Vertx
+ * instance).
+ */
+ private final class SlowComponent extends DefaultComponent {
+ private final Map<String, ServiceStatus> statusAtCreate = new
ConcurrentHashMap<>();
+ private volatile Object client;
+ private volatile String startedBy;
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ startedBy = Thread.currentThread().getName();
+ starting.countDown();
+ assertTrue(release.await(5, TimeUnit.SECONDS));
+ client = new Object();
+ }
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining,
Map<String, Object> parameters) {
+ statusAtCreate.put(remaining, getStatus());
+ created.countDown();
+ // the endpoint needs the client of the component
+ String name = remaining + "@" + client.hashCode();
+ return new DefaultEndpoint(uri, this) {
+ @Override
+ public Producer createProducer() {
+ throw new UnsupportedOperationException(name);
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) {
+ throw new UnsupportedOperationException(name);
+ }
+ };
+ }
+ }
+}
