fix: create the PollingEventSource timer on start and make it a daemon (#3523)

`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.

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

Author: csviri

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