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 df4fcf2cb0a1 CAMEL-26334: camel-once - the consumer releases its
thread when its route stops, and fires again after a restart (#28757)
df4fcf2cb0a1 is described below
commit df4fcf2cb0a1bdef8c5deb92ab129ad50091ea32
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 31 14:59:04 2026 +0530
CAMEL-26334: camel-once - the consumer releases its thread when its route
stops, and fires again after a restart (#28757)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/component/once/OnceConsumer.java | 43 ++++--
.../component/once/OnceConsumerLifecycleTest.java | 167 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 12 ++
3 files changed, 209 insertions(+), 13 deletions(-)
diff --git
a/components/camel-once/src/main/java/org/apache/camel/component/once/OnceConsumer.java
b/components/camel-once/src/main/java/org/apache/camel/component/once/OnceConsumer.java
index c4d29d7d8eaf..87440e5fdec6 100644
---
a/components/camel-once/src/main/java/org/apache/camel/component/once/OnceConsumer.java
+++
b/components/camel-once/src/main/java/org/apache/camel/component/once/OnceConsumer.java
@@ -17,8 +17,9 @@
package org.apache.camel.component.once;
import java.io.InputStream;
-import java.util.Timer;
-import java.util.TimerTask;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
@@ -38,30 +39,41 @@ public class OnceConsumer extends DefaultConsumer
implements StartupListener {
private final CamelContext camelContext;
private final OnceEndpoint endpoint;
- private final Timer timer;
- private final TimerTask task;
+ private final Runnable task;
+ // the thread that fires the route is created when the consumer starts and
released when it stops: it must not
+ // outlive the route (one thread per route that was ever created) nor keep
the JVM running
+ private volatile ScheduledExecutorService executor;
+ private volatile ScheduledFuture<?> future;
private volatile boolean scheduled;
public OnceConsumer(OnceEndpoint endpoint, Processor processor) {
super(endpoint, processor);
this.camelContext = endpoint.getCamelContext();
this.endpoint = endpoint;
- this.timer = new Timer(endpoint.getName());
- this.task = new OnceTimerTask();
+ this.task = new OnceTask();
}
@Override
protected void doStart() throws Exception {
super.doStart();
+ executor =
camelContext.getExecutorServiceManager().newSingleThreadScheduledExecutor(this,
endpoint.getName());
if (!scheduled && endpoint.getCamelContext().getStatus().isStarted()) {
- scheduleTask(task, timer);
+ scheduleTask();
}
}
@Override
protected void doStop() throws Exception {
- if (task != null) {
- task.cancel();
+ ScheduledFuture<?> f = future;
+ if (f != null) {
+ // an exchange that is being routed is not interrupted
+ f.cancel(false);
+ future = null;
+ }
+ ScheduledExecutorService e = executor;
+ if (e != null) {
+ camelContext.getExecutorServiceManager().shutdown(e);
+ executor = null;
}
scheduled = false;
super.doStop();
@@ -70,18 +82,23 @@ public class OnceConsumer extends DefaultConsumer
implements StartupListener {
@Override
public void onCamelContextStarted(CamelContext context, boolean
alreadyStarted) throws Exception {
if (!scheduled) {
- scheduleTask(task, timer);
+ scheduleTask();
}
}
- protected void scheduleTask(TimerTask task, Timer timer) {
+ protected void scheduleTask() {
+ ScheduledExecutorService e = executor;
+ if (e == null) {
+ // not started
+ return;
+ }
long delay = Math.max(0, endpoint.getDelay());
LOG.debug("Scheduled once after: {} mills for task: {} ", delay, task);
- timer.schedule(task, delay);
+ future = e.schedule(task, delay, TimeUnit.MILLISECONDS);
scheduled = true;
}
- private class OnceTimerTask extends TimerTask {
+ private class OnceTask implements Runnable {
@Override
public void run() {
diff --git
a/components/camel-once/src/test/java/org/apache/camel/component/once/OnceConsumerLifecycleTest.java
b/components/camel-once/src/test/java/org/apache/camel/component/once/OnceConsumerLifecycleTest.java
new file mode 100644
index 000000000000..0e5872786da8
--- /dev/null
+++
b/components/camel-once/src/test/java/org/apache/camel/component/once/OnceConsumerLifecycleTest.java
@@ -0,0 +1,167 @@
+/*
+ * 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.component.once;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The once consumer releases the thread that fires it when its route stops,
and fires again when the route is started
+ * again.
+ */
+class OnceConsumerLifecycleTest {
+
+ @Test
+ void theThreadEndsWhenTheCamelContextStops() throws Exception {
+ String name = "onceContextStop" + System.nanoTime();
+ CountDownLatch fired = new CountDownLatch(1);
+ try (CamelContext context = new DefaultCamelContext()) {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("once:" + name + "?delay=0").process(e ->
fired.countDown());
+ }
+ });
+ context.start();
+ assertTrue(fired.await(10, TimeUnit.SECONDS), "the route did not
fire");
+
+ context.stop();
+
+ assertEquals(List.of(), threadsStillAliveOf(name), "threads of the
stopped once consumer");
+ }
+ }
+
+ @Test
+ void theThreadEndsWhenTheRouteIsRemovedBeforeItFired() throws Exception {
+ String name = "onceRemoved" + System.nanoTime();
+ try (CamelContext context = new DefaultCamelContext()) {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("once:" + name +
"?delay=3600000").routeId("r").to("log:never");
+ }
+ });
+ context.start();
+
+ context.getRouteController().stopRoute("r");
+ context.removeRoute("r");
+
+ assertEquals(List.of(), threadsStillAliveOf(name), "threads of the
removed once consumer");
+ }
+ }
+
+ @Test
+ void noThreadIsLeftPerRouteThatWasAddedAndRemoved() throws Exception {
+ String prefix = "onceMany" + System.nanoTime() + "x";
+ try (CamelContext context = new DefaultCamelContext()) {
+ context.start();
+ for (int i = 0; i < 20; i++) {
+ String routeId = "r" + i;
+ CountDownLatch fired = new CountDownLatch(1);
+ int n = i;
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("once:" + prefix + n +
"?delay=0").routeId(routeId).process(e -> fired.countDown());
+ }
+ });
+ assertTrue(fired.await(10, TimeUnit.SECONDS), "route " +
routeId + " did not fire");
+ context.getRouteController().stopRoute(routeId);
+ context.removeRoute(routeId);
+ }
+
+ assertEquals(List.of(), threadsStillAliveOf(prefix), "threads of
the removed once consumers");
+ }
+ }
+
+ @Test
+ void aRouteThatIsStartedAgainFiresAgain() throws Exception {
+ String name = "onceRestart" + System.nanoTime();
+ AtomicInteger count = new AtomicInteger();
+ Semaphore fired = new Semaphore(0);
+ try (CamelContext context = new DefaultCamelContext()) {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("once:" + name + "?delay=0").routeId("r").process(e
-> {
+ count.incrementAndGet();
+ fired.release();
+ });
+ }
+ });
+ context.start();
+ assertTrue(fired.tryAcquire(10, TimeUnit.SECONDS), "the route did
not fire");
+
+ context.getRouteController().stopRoute("r");
+ context.getRouteController().startRoute("r");
+
+ assertTrue(fired.tryAcquire(10, TimeUnit.SECONDS), "the started
again route did not fire");
+ assertEquals(2, count.get());
+ }
+ }
+
+ @Test
+ void aRouteFiresAgainWhenTheCamelContextIsStartedAgain() throws Exception {
+ String name = "onceContextRestart" + System.nanoTime();
+ Semaphore fired = new Semaphore(0);
+ try (CamelContext context = new DefaultCamelContext()) {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("once:" + name + "?delay=0").process(e ->
fired.release());
+ }
+ });
+ context.start();
+ assertTrue(fired.tryAcquire(10, TimeUnit.SECONDS), "the route did
not fire");
+
+ context.stop();
+ context.start();
+
+ assertTrue(fired.tryAcquire(10, TimeUnit.SECONDS), "the route did
not fire after the CamelContext restart");
+ }
+ }
+
+ /**
+ * The threads whose name contains the given text and that are still alive
once they had up to 10 seconds in all to
+ * end (a thread that is shut down ends at once, a leaked timer thread
never does).
+ */
+ private static List<String> threadsStillAliveOf(String text) throws
InterruptedException {
+ List<String> alive = new ArrayList<>();
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+ for (Thread t : Thread.getAllStackTraces().keySet()) {
+ if (t.getName().contains(text)) {
+ t.join(Math.max(1, TimeUnit.NANOSECONDS.toMillis(deadline -
System.nanoTime())));
+ if (t.isAlive()) {
+ alive.add(t.getName() + (t.isDaemon() ? " (daemon)" : "
(non-daemon)"));
+ }
+ }
+ }
+ return alive;
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index bd8443700b2d..d298864d2e54 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -4258,3 +4258,15 @@ To find the dependencies, `camel export` boots the
integration silently, as befo
exported project: such an export can now have more (or other) dependencies.
The security policy of the profile (such
as `prod`, which fails on an insecure option) does not fail the export, nor
does `camel.security.policy=fail`: the
export only warns. Without `--profile` the dependencies are the same as
before.
+
+=== camel-once - the consumer fires on a Camel thread pool
+
+The once consumer now fires its route on a single thread scheduled executor
that it creates through the
+`ExecutorServiceManager` when the route starts and shuts down when the route
stops, so the thread is named
+`Camel (<context>) thread #N - <name>` instead of `<name>`. Before, each once
consumer created a non-daemon
+`java.util.Timer` thread when it was created, and that thread stayed alive
after its route was stopped and after the
+CamelContext was stopped (after its route was removed, until the consumer was
garbage collected). A once route that is
+stopped and started again now fires again, and so does a once route after the
CamelContext is stopped and started
+again; before, the start failed with `IllegalStateException: Task already
scheduled or cancelled`.
+
+The protected method `OnceConsumer.scheduleTask(TimerTask, Timer)` is replaced
by `scheduleTask()`.
