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

[csviri]

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.

Part of #3517

[Copilot]

Pull request overview

This PR fixes PollingEventSource lifecycle and JVM-exit behavior by moving java.util.Timer creation to start() (as a daemon) and clearing it in stop(), aligning its lifecycle with supported operator restart semantics and with TimerEventSource.

Changes:

• Create the PollingEventSource timer on start() (daemon thread) and null it out on stop().
• Add regression tests covering restart-after-stop and verifying the timer thread is daemon.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.

File	Description
operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSource.java	Timer lifecycle moved to start()/stop() and daemonized to support restart and avoid blocking JVM shutdown.
operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSourceTest.java	Adds regression coverage for restart behavior and daemon timer thread behavior.

[csviri]

LGTM

[Copilot]

Pull request overview

Copilot reviewed 2 out of 2 changed files in this pull request and generated no new comments.

Suppressed comments (2)

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

• stop() can race with a concurrent start() (e.g., dynamic deregistration/registration) because neither method is synchronized; this can lead to stop() cancelling the newly-created timer after start() assigns it. Synchronizing stop() (together with start()) makes lifecycle transitions atomic for this event source instance.

  public void stop() throws OperatorException {

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

• start() is guarded by timer != null, but timer is only assigned after getStateAndFillCache(). EventSourceManager#dynamicallyRegisterEventSource calls eventSource.start() outside its synchronized block, so concurrent start() calls on the same instance are possible; with the current ordering both callers can observe timer == null and end up creating/scheduling multiple timers/tasks (and potentially leaking the first timer when the field is overwritten). Consider guarding with isRunning() (set by super.start()) and initializing the timer before the initial cache fill so the second concurrent call returns immediately.

This issue also appears on line 116 of the same file.

  public void start() throws OperatorException {
    if (timer != null) {
      return;
    }
    super.start();