Skip to content

Commit 44e53a1

Browse files
committed
fix: create the PollingEventSource timer on start and make it a daemon
`PollingEventSource` held its timer in a final field initialised at construction: private final Timer timer = new Timer(); Two problems follow from that. The event source cannot be restarted. `stop()` calls `timer.cancel()` and a cancelled `java.util.Timer` cannot be reused, so a subsequent `start()` fails with `IllegalStateException: Timer already cancelled`. Restart is a supported lifecycle - `Operator.stop()` / `start()` recreates the thread pools for exactly this reason, and `TimerEventSource` creates a new `Timer` inside `start()`. The timer thread is not a daemon and is created eagerly. Merely constructing a `PollingEventSource` therefore starts a non-daemon thread that keeps the JVM from exiting, even if the event source is never started, and it outlives an operator that is stopped without stopping its event sources. `TimerEventSource` uses `new Timer(true)`. The timer is now created in `start()` as a daemon and cleared in `stop()`, matching `TimerEventSource`. Adds regression tests for restart and for the daemon flag; the restart one fails with `IllegalStateException: Timer already cancelled` without this change. Note: `PerResourcePollingEventSource` has a related restart limitation because `stop()` calls `shutdownNow()` on the `ScheduledExecutorService` from its configuration. That executor is supplied by the caller, so changing its ownership semantics is a separate discussion and is left out of this change.
1 parent 67c47c7 commit 44e53a1

2 files changed

Lines changed: 34 additions & 2 deletions

File tree

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSource.java

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -62,7 +62,7 @@ public class PollingEventSource<R, P extends HasMetadata, ID>
6262

6363
private static final Logger log = LoggerFactory.getLogger(PollingEventSource.class);
6464

65-
private final Timer timer = new Timer();
65+
private Timer timer;
6666
private final GenericResourceFetcher<R> genericResourceFetcher;
6767
private final Duration period;
6868
private final AtomicBoolean healthy = new AtomicBoolean(true);
@@ -76,6 +76,9 @@ public PollingEventSource(Class<R> resourceClass, PollingConfiguration<R, ID> co
7676
@Override
7777
public void start() throws OperatorException {
7878
super.start();
79+
// a cancelled Timer cannot be reused, so a fresh one is created on every start; it is a daemon
80+
// thread so that it never keeps the JVM alive
81+
timer = new Timer(true);
7982
getStateAndFillCache();
8083
timer.schedule(
8184
new TimerTask() {
@@ -111,7 +114,10 @@ public interface GenericResourceFetcher<R> {
111114
@Override
112115
public void stop() throws OperatorException {
113116
super.stop();
114-
timer.cancel();
117+
if (timer != null) {
118+
timer.cancel();
119+
timer = null;
120+
}
115121
}
116122

117123
@Override

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSourceTest.java

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,32 @@ public void setup() {
6060
setUpSource(pollingEventSource, false);
6161
}
6262

63+
@Test
64+
void canBeRestartedAfterStop() throws InterruptedException {
65+
when(resourceFetcher.fetchResources()).thenReturn(testResponseWithTwoValues());
66+
pollingEventSource.start();
67+
Thread.sleep(DEFAULT_WAIT_PERIOD);
68+
pollingEventSource.stop();
69+
70+
// a cancelled java.util.Timer cannot be reused, a new one has to be created on start
71+
pollingEventSource.start();
72+
Thread.sleep(DEFAULT_WAIT_PERIOD);
73+
74+
assertThat(pollingEventSource.getStatus()).isEqualTo(Status.HEALTHY);
75+
}
76+
77+
@Test
78+
void timerThreadIsADaemonSoItDoesNotKeepTheJvmAlive() throws InterruptedException {
79+
when(resourceFetcher.fetchResources()).thenReturn(testResponseWithTwoValues());
80+
pollingEventSource.start();
81+
Thread.sleep(DEFAULT_WAIT_PERIOD);
82+
83+
assertThat(Thread.getAllStackTraces().keySet())
84+
.filteredOn(t -> t.getName().startsWith("Timer-"))
85+
.isNotEmpty()
86+
.allMatch(Thread::isDaemon);
87+
}
88+
6389
@Test
6490
void pollsAndProcessesEvents() throws InterruptedException {
6591
when(resourceFetcher.fetchResources()).thenReturn(testResponseWithTwoValues());

0 commit comments

Comments
 (0)