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 9be82b6130c4 CAMEL-26537: camel-core - TimerListenerManager: adding or
removing a listener while the timer runs must not stop the timer (#29121)
9be82b6130c4 is described below
commit 9be82b6130c4c8491654b0f4e8559119315316cb
Author: allthingssecurity <[email protected]>
AuthorDate: Wed Nov 4 13:43:57 2026 +0530
CAMEL-26537: camel-core - TimerListenerManager: adding or removing a
listener while the timer runs must not stop the timer (#29121)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/impl/engine/PeriodTaskSchedulerTest.java | 65 ++++++++++++
.../camel/support/TimerListenerManagerTest.java | 109 +++++++++++++++++++++
.../apache/camel/support/TimerListenerManager.java | 5 +-
3 files changed, 177 insertions(+), 2 deletions(-)
diff --git
a/core/camel-core/src/test/java/org/apache/camel/impl/engine/PeriodTaskSchedulerTest.java
b/core/camel-core/src/test/java/org/apache/camel/impl/engine/PeriodTaskSchedulerTest.java
index f1cff294801f..4a167462f293 100644
---
a/core/camel-core/src/test/java/org/apache/camel/impl/engine/PeriodTaskSchedulerTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/impl/engine/PeriodTaskSchedulerTest.java
@@ -16,7 +16,9 @@
*/
package org.apache.camel.impl.engine;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.camel.ContextTestSupport;
@@ -128,4 +130,67 @@ public class PeriodTaskSchedulerTest extends
ContextTestSupport {
Awaitility.waitAtMost(5, TimeUnit.SECONDS).until(() -> counter.get()
== 1 && taskCounter.get() == 10);
}
+ @Test
+ public void testSchedulePeriodTaskWhileTimerRuns() throws Exception {
+ final AtomicBoolean first = new AtomicBoolean(true);
+ final CountDownLatch entered = new CountDownLatch(1);
+ final CountDownLatch released = new CountDownLatch(1);
+ final AtomicInteger secondCounter = new AtomicInteger();
+ final AtomicInteger addedCounter = new AtomicInteger();
+
+ PeriodTaskScheduler scheduler =
PluginHelper.getPeriodTaskScheduler(context);
+ if (scheduler instanceof TimerListenerManager timerListenerManager) {
+ // speedup unit test
+ timerListenerManager.setInterval(10);
+ }
+ // the first task blocks the timer thread in its first run (like a
task calling a remote service)
+ scheduler.schedulePeriodTask(() -> {
+ if (first.compareAndSet(true, false)) {
+ entered.countDown();
+ try {
+ released.await(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }, 1);
+ scheduler.schedulePeriodTask(secondCounter::incrementAndGet, 1);
+ context.start();
+ try {
+ Assertions.assertTrue(entered.await(5, TimeUnit.SECONDS), "The
first task should run");
+ // a task scheduled while the timer runs the first task (like a
consumer started at runtime)
+ scheduler.schedulePeriodTask(addedCounter::incrementAndGet, 1);
+ released.countDown();
+
+ // the timer must keep running the tasks
+ Awaitility.waitAtMost(5, TimeUnit.SECONDS).until(() ->
secondCounter.get() > 1 && addedCounter.get() > 1);
+ } finally {
+ released.countDown();
+ }
+ }
+
+ @Test
+ public void testSchedulePeriodTaskFromRunningTask() {
+ final AtomicBoolean first = new AtomicBoolean(true);
+ final AtomicInteger secondCounter = new AtomicInteger();
+ final AtomicInteger addedCounter = new AtomicInteger();
+
+ PeriodTaskScheduler scheduler =
PluginHelper.getPeriodTaskScheduler(context);
+ if (scheduler instanceof TimerListenerManager timerListenerManager) {
+ // speedup unit test
+ timerListenerManager.setInterval(10);
+ }
+ // the first task schedules a task from its run, on the timer thread
(like a vault refresh task that
+ // reloads the routes, whose consumers schedule their own tasks when
they start)
+ scheduler.schedulePeriodTask(() -> {
+ if (first.compareAndSet(true, false)) {
+ scheduler.schedulePeriodTask(addedCounter::incrementAndGet, 1);
+ }
+ }, 1);
+ scheduler.schedulePeriodTask(secondCounter::incrementAndGet, 1);
+ context.start();
+
+ // the timer must keep running the tasks
+ Awaitility.waitAtMost(5, TimeUnit.SECONDS).until(() ->
secondCounter.get() > 1 && addedCounter.get() > 1);
+ }
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/support/TimerListenerManagerTest.java
b/core/camel-core/src/test/java/org/apache/camel/support/TimerListenerManagerTest.java
index 5340eb10ee02..81345bd83b30 100644
---
a/core/camel-core/src/test/java/org/apache/camel/support/TimerListenerManagerTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/support/TimerListenerManagerTest.java
@@ -20,6 +20,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.TimerListener;
@@ -54,6 +55,114 @@ public class TimerListenerManagerTest extends
ContextTestSupport {
executor.shutdown();
}
+ @Test
+ public void testAddListenerWhileTimerRuns() throws Exception {
+ TimerListenerManager manager = new TimerListenerManager();
+ manager.setCamelContext(context);
+ manager.setInterval(10);
+ BlockingListener blocking = new BlockingListener();
+ CountingListener next = new CountingListener(2);
+ manager.addTimerListener(blocking);
+ manager.addTimerListener(next);
+ manager.start();
+ try {
+ // the timer thread is in onTimer of the first listener, the
second listener is still to be called
+ assertTrue(blocking.awaitEntered(), "The timer should call the
first listener");
+ CountingListener added = new CountingListener(1);
+ manager.addTimerListener(added);
+ blocking.release();
+
+ // the timer must keep running and call every listener on the next
runs
+ assertTrue(next.await(), "The timer should keep calling the second
listener");
+ assertTrue(added.await(), "The timer should call the listener
added while it was running");
+ } finally {
+ blocking.release();
+ manager.shutdown();
+ }
+ }
+
+ @Test
+ public void testRemoveListenerWhileTimerRuns() throws Exception {
+ TimerListenerManager manager = new TimerListenerManager();
+ manager.setCamelContext(context);
+ manager.setInterval(10);
+ BlockingListener blocking = new BlockingListener();
+ // the run in progress may still call the removed listener once
+ TimerListener removed = new CountingListener(1);
+ CountingListener last = new CountingListener(2);
+ manager.addTimerListener(blocking);
+ manager.addTimerListener(removed);
+ manager.addTimerListener(last);
+ manager.start();
+ try {
+ assertTrue(blocking.awaitEntered(), "The timer should call the
first listener");
+ manager.removeTimerListener(removed);
+ blocking.release();
+
+ // the timer must keep running and call the remaining listeners on
the next runs
+ assertTrue(last.await(), "The timer should keep calling the
remaining listeners");
+ assertTrue(blocking.awaitCalledAgain(), "The timer should keep
calling the first listener");
+ } finally {
+ blocking.release();
+ manager.shutdown();
+ }
+ }
+
+ /**
+ * Blocks the timer thread in its first call until released.
+ */
+ private static class BlockingListener implements TimerListener {
+
+ private final AtomicBoolean first = new AtomicBoolean(true);
+ private final CountDownLatch entered = new CountDownLatch(1);
+ private final CountDownLatch released = new CountDownLatch(1);
+ private final CountDownLatch calledAgain = new CountDownLatch(1);
+
+ @Override
+ public void onTimer() {
+ if (first.compareAndSet(true, false)) {
+ entered.countDown();
+ try {
+ released.await(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ } else {
+ calledAgain.countDown();
+ }
+ }
+
+ boolean awaitEntered() throws InterruptedException {
+ return entered.await(5, TimeUnit.SECONDS);
+ }
+
+ boolean awaitCalledAgain() throws InterruptedException {
+ return calledAgain.await(5, TimeUnit.SECONDS);
+ }
+
+ void release() {
+ released.countDown();
+ }
+ }
+
+ private static class CountingListener implements TimerListener {
+
+ private final CountDownLatch latch;
+
+ CountingListener(int calls) {
+ this.latch = new CountDownLatch(calls);
+ }
+
+ @Override
+ public void onTimer() {
+ latch.countDown();
+ }
+
+ boolean await() throws InterruptedException {
+ return latch.await(5, TimeUnit.SECONDS);
+ }
+ }
+
private static class MyTask implements TimerListener {
private final CountDownLatch latch = new CountDownLatch(1);
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/TimerListenerManager.java
b/core/camel-support/src/main/java/org/apache/camel/support/TimerListenerManager.java
index 58a25324f44e..13c817194823 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/TimerListenerManager.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/TimerListenerManager.java
@@ -17,8 +17,8 @@
package org.apache.camel.support;
import java.util.Collections;
-import java.util.LinkedHashSet;
import java.util.Set;
+import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
@@ -44,7 +44,8 @@ import org.slf4j.LoggerFactory;
public class TimerListenerManager extends ServiceSupport implements Runnable,
CamelContextAware, StaticService {
private static final Logger LOG =
LoggerFactory.getLogger(TimerListenerManager.class);
- private final Set<TimerListener> listeners = new LinkedHashSet<>();
+ // listeners are added and removed by other threads while the timer thread
iterates them
+ private final Set<TimerListener> listeners = new CopyOnWriteArraySet<>();
private CamelContext camelContext;
private ScheduledExecutorService executorService;
private volatile ScheduledFuture<?> task;
