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);
@@ -75,8 +75,12 @@ public PollingEventSource(Class<R> resourceClass, PollingConfiguration<R, ID> 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<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,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());