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

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;
     }
   }
 

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);
+  }
+}