diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSource.java index f5e5a79430..5c85c24f90 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSource.java @@ -62,7 +62,7 @@ public class PollingEventSource private static final Logger log = LoggerFactory.getLogger(PollingEventSource.class); - private final Timer timer = new Timer(); + private Timer timer; private final GenericResourceFetcher genericResourceFetcher; private final Duration period; private final AtomicBoolean healthy = new AtomicBoolean(true); @@ -75,8 +75,12 @@ public PollingEventSource(Class resourceClass, PollingConfiguration co @Override public void start() throws OperatorException { + if (timer != null) { + return; + } super.start(); getStateAndFillCache(); + timer = new Timer(true); timer.schedule( new TimerTask() { @Override @@ -111,7 +115,10 @@ public interface GenericResourceFetcher { @Override public void stop() throws OperatorException { super.stop(); - timer.cancel(); + if (timer != null) { + timer.cancel(); + timer = null; + } } @Override diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSourceTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSourceTest.java index 12dbf17be7..92400df4de 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSourceTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSourceTest.java @@ -60,6 +60,38 @@ public void setup() { setUpSource(pollingEventSource, false); } + @Test + void canBeRestartedAfterStop() throws InterruptedException { + when(resourceFetcher.fetchResources()).thenReturn(testResponseWithTwoValues()); + pollingEventSource.start(); + Thread.sleep(DEFAULT_WAIT_PERIOD); + pollingEventSource.stop(); + + // a cancelled java.util.Timer cannot be reused, a new one has to be created on start + pollingEventSource.start(); + Thread.sleep(DEFAULT_WAIT_PERIOD); + + assertThat(pollingEventSource.getStatus()).isEqualTo(Status.HEALTHY); + } + + @Test + void timerThreadIsADaemonSoItDoesNotKeepTheJvmAlive() throws InterruptedException { + when(resourceFetcher.fetchResources()).thenReturn(testResponseWithTwoValues()); + + var threadsBeforeStart = Thread.getAllStackTraces().keySet(); + + pollingEventSource.start(); + Thread.sleep(DEFAULT_WAIT_PERIOD); + + var newTimerThreads = + Thread.getAllStackTraces().keySet().stream() + .filter(t -> t.getName().startsWith("Timer-")) + .filter(t -> !threadsBeforeStart.contains(t)) + .toList(); + + assertThat(newTimerThreads).isNotEmpty().allMatch(Thread::isDaemon); + } + @Test void pollsAndProcessesEvents() throws InterruptedException { when(resourceFetcher.fetchResources()).thenReturn(testResponseWithTwoValues());