From 7ea836b66154d71e607cfcb95be8cdddc6a3f2ea Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Attila=20M=C3=A9sz=C3=A1ros?= Date: Wed, 29 Jul 2026 18:22:16 +0200 Subject: [PATCH] fix: always release executors and reset state when stopping the executor manager `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. --- .../api/config/ExecutorServiceManager.java | 16 ++++-- .../config/ExecutorServiceManagerTest.java | 56 +++++++++++++++++++ 2 files changed, 67 insertions(+), 5 deletions(-) create mode 100644 operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManagerTest.java diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManager.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManager.java index 58928b1302..081514c443 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManager.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManager.java @@ -143,20 +143,26 @@ public synchronized void start(ConfigurationService configurationService) { } public synchronized void stop(Duration gracefulShutdownTimeout) { + var parallelExec = Executors.newFixedThreadPool(4); try { log.debug("Closing executor"); - var parallelExec = Executors.newFixedThreadPool(3); parallelExec.invokeAll( List.of( shutdown(executor, gracefulShutdownTimeout), shutdown(workflowExecutor, gracefulShutdownTimeout), - shutdown(cachingExecutorService, gracefulShutdownTimeout))); - workflowExecutor = null; - parallelExec.shutdownNow(); - started = false; + shutdown(cachingExecutorService, gracefulShutdownTimeout), + shutdown(scheduledExecutorService, gracefulShutdownTimeout))); } catch (InterruptedException e) { log.debug("Exception closing executor: {}", e.getLocalizedMessage()); Thread.currentThread().interrupt(); + } finally { + // this has to happen even if we were interrupted, otherwise the helper pool leaks its + // (non-daemon) threads, and leaving started == true would make a subsequent start() a no-op, + // silently leaving the manager with already terminated executors + parallelExec.shutdownNow(); + workflowExecutor = null; + scheduledExecutorService = null; + started = false; } } diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManagerTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManagerTest.java new file mode 100644 index 0000000000..c31ea4d1f7 --- /dev/null +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManagerTest.java @@ -0,0 +1,56 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.api.config; + +import java.time.Duration; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +class ExecutorServiceManagerTest { + + private static final Duration SHUTDOWN_TIMEOUT = Duration.ofMillis(100); + + @Test + void stopShutsDownTheScheduledExecutorService() { + ConfigurationService configurationService = new BaseConfigurationService(); + var manager = configurationService.getExecutorServiceManager(); + var scheduled = manager.scheduledExecutorService(); + assertThat(scheduled.isShutdown()).isFalse(); + + manager.stop(SHUTDOWN_TIMEOUT); + + assertThat(scheduled.isShutdown()).isTrue(); + } + + @Test + 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); + } +}