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..e61a7a89d1 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); @@ -76,6 +76,9 @@ public PollingEventSource(Class resourceClass, PollingConfiguration co @Override public void start() throws OperatorException { super.start(); + // a cancelled Timer cannot be reused, so a fresh one is created on every start; it is a daemon + // thread so that it never keeps the JVM alive + timer = new Timer(true); getStateAndFillCache(); timer.schedule( new TimerTask() { @@ -111,7 +114,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..744d979567 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,32 @@ 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()); + pollingEventSource.start(); + Thread.sleep(DEFAULT_WAIT_PERIOD); + + assertThat(Thread.getAllStackTraces().keySet()) + .filteredOn(t -> t.getName().startsWith("Timer-")) + .isNotEmpty() + .allMatch(Thread::isDaemon); + } + @Test void pollsAndProcessesEvents() throws InterruptedException { when(resourceFetcher.fetchResources()).thenReturn(testResponseWithTwoValues());