fix: always release executors and reset state when stopping the executor manager

[csviri]

ExecutorServiceManager.stop had three problems, all on the interrupted
path or affecting the scheduled executor.

scheduledExecutorService was never shut down. It is created on every
start() and exposed through a public accessor, but stop() only shut
down the reconcile, workflow and caching executors, so the pool (and any
non-daemon threads a caller created through the accessor) outlived the
operator and leaked again on every restart.

The helper pool leaked when interrupted. Executors.newFixedThreadPool(3)
was created inside the try and only shut down on the success path, so an
InterruptedException from invokeAll left three non-daemon threads
behind - in a shutdown path, where they then keep the JVM alive.

started was not reset when interrupted. It was only set to false on the
success path, so after an interrupted stop() the executors were already
shut down but start() would see started == true and do nothing. The
operator then looked started while every execute on the terminated
reconcile executor failed with RejectedExecutionException.

Moves the cleanup into a finally, includes the scheduled executor in the
graceful shutdown, and clears both nullable references there.

Adds regression tests for the scheduled executor shutdown and for
restartability; the former fails without this change.

Part of #3517

[Copilot]

Pull request overview

This PR fixes shutdown/restart correctness in ExecutorServiceManager.stop() to prevent executor/thread leaks and ensure the manager can be restarted cleanly after a stop (including interrupted stop paths).

Changes:

• Shut down the scheduledExecutorService alongside other executors during stop().
• Move helper-pool shutdown and state reset into a finally block so it runs even when interrupted.
• Add regression tests for scheduled-executor shutdown and manager restartability.

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/api/config/ExecutorServiceManager.java	Ensures shutdown logic runs reliably (including scheduled executor) and resets state in finally.
operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManagerTest.java	Adds regression tests covering scheduled executor shutdown and restartability after stop.

Comments suppressed due to low confidence (1)

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManagerTest.java:55

• As written, if any of the post-restart assertions fail, the test will exit before the final manager.stop(...) call and may leave non-daemon executor threads running. Wrapping the restart assertions in a try/finally ensures cleanup even when the test fails.

    manager.stop(SHUTDOWN_TIMEOUT);
    manager.start(configurationService);

    // start() is a no-op unless stop() reset the started flag, which would leave the manager
    // handing out already terminated executors
    assertThat(manager.reconcileExecutorService().isShutdown()).isFalse();
    assertThat(manager.cachingExecutorService().isShutdown()).isFalse();
    assertThat(manager.scheduledExecutorService().isShutdown()).isFalse();

    manager.stop(SHUTDOWN_TIMEOUT);
  }

[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/test/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManagerTest.java:58

• This test only stops the manager on the success path. If an assertion fails, the executors created by BaseConfigurationService can remain alive and keep the JVM running, which can cascade into unrelated test failures/timeouts. Wrap the body in a try/finally and always call stop() in the finally block (similar to the other test in this class).

  void canBeRestartedAfterStop() {
    ConfigurationService configurationService = new BaseConfigurationService();
    var manager = configurationService.getExecutorServiceManager();

    manager.stop(SHUTDOWN_TIMEOUT);
    manager.start(configurationService);

    // start() is a no-op unless stop() reset the started flag, which would leave the manager
    // handing out already terminated executors
    assertThat(manager.reconcileExecutorService().isShutdown()).isFalse();
    assertThat(manager.cachingExecutorService().isShutdown()).isFalse();
    assertThat(manager.scheduledExecutorService().isShutdown()).isFalse();

    manager.stop(SHUTDOWN_TIMEOUT);
  }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManager.java:157

• If stop() is interrupted while waiting for invokeAll(), the JDK cancels unfinished tasks, which can leave some target executors (including the scheduled executor) not shut down even though stop() returns. Consider performing a best-effort shutdownNow() of the managed executors in the catch block so interruption can’t prevent their shutdown.

    } catch (InterruptedException e) {
      log.debug("Exception closing executor: {}", e.getLocalizedMessage());
      Thread.currentThread().interrupt();