fix: create the PollingEventSource timer on start and make it a daemon

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

@@ -62,7 +62,7 @@ public class PollingEventSource<R, P extends HasMetadata, ID>
 
   private static final Logger log = LoggerFactory.getLogger(PollingEventSource.class);
 
-  private final Timer timer = new Timer();
+  private Timer timer;
   private final GenericResourceFetcher<R> genericResourceFetcher;
   private final Duration period;
   private final AtomicBoolean healthy = new AtomicBoolean(true);
@@ -76,6 +76,9 @@ public PollingEventSource(Class<R> resourceClass, PollingConfiguration<R, ID> 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<R> {
   @Override
   public void stop() throws OperatorException {
     super.stop();
-    timer.cancel();
+    if (timer != null) {
+      timer.cancel();
+      timer = null;
+    }
   }
 
   @Override

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());