feat: use only one, configurable ExecutorService

Fixes #540

Author: metacosm

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/Operator.java

@@ -6,6 +6,7 @@
 import java.util.Collections;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -102,6 +103,18 @@ public void close() {
     log.info(
         "Operator SDK {} is shutting down...", configurationService.getVersion().getSdkVersion());
 
+    try {
+      log.debug("Closing executor");
+      final var executor = configurationService.getExecutorService();
+      executor.shutdown();
+      if (!executor.awaitTermination(configurationService.getTerminationTimeoutSeconds(),
+          TimeUnit.SECONDS)) {
+        executor.shutdownNow(); // if we timed out, waiting, cancel everything
+      }
+    } catch (InterruptedException e) {
+      log.debug("Exception closing executor: {}", e.getLocalizedMessage());
+    }
+
     controllers.close();
 
     k8sClient.close();

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java

@@ -8,6 +8,7 @@
 import io.javaoperatorsdk.operator.api.ResourceController;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.concurrent.ExecutorService;
 
 /** An interface from which to retrieve configuration information. */
 public interface ConfigurationService {
@@ -102,4 +103,8 @@ default int getTerminationTimeoutSeconds() {
   default Metrics getMetrics() {
     return Metrics.NOOP;
   }
+
+  default ExecutorService getExecutorService() {
+    return ExecutorServiceProducer.getExecutor(concurrentReconciliationThreads());
+  }
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceProducer.java

@@ -0,0 +1,26 @@
+package io.javaoperatorsdk.operator.api.config;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.atomic.AtomicReference;
+
+class ExecutorServiceProducer {
+
+  private final static AtomicReference<ScheduledThreadPoolExecutor> executor =
+      new AtomicReference<>();
+
+  static ExecutorService getExecutor(int threadPoolSize) {
+    final var gotSet =
+        executor.compareAndSet(null, new ScheduledThreadPoolExecutor(threadPoolSize));
+    final var result = executor.get();
+    if (!gotSet) {
+      // check that we didn't try to change the pool size
+      if (result.getCorePoolSize() != threadPoolSize) {
+        throw new IllegalArgumentException(
+            "Cannot change the ExecutorService's thread pool size once set! Was "
+                + result.getCorePoolSize() + ", attempted to retrieve " + threadPoolSize);
+      }
+    }
+    return result;
+  }
+}

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/DefaultEventHandler.java

@@ -5,6 +5,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.ReentrantLock;
@@ -44,61 +45,37 @@ public void failedEvent(String uid, Event event) {}
 
   private final EventBuffer eventBuffer;
   private final Set<String> underProcessing = new HashSet<>();
-  private final ScheduledThreadPoolExecutor executor;
   private final EventDispatcher<R> eventDispatcher;
   private final Retry retry;
   private final Map<String, RetryExecution> retryState = new HashMap<>();
+  private final ExecutorService executor;
   private final String controllerName;
-  private final int terminationTimeout;
   private final ReentrantLock lock = new ReentrantLock();
   private DefaultEventSourceManager<R> eventSourceManager;
 
   public DefaultEventHandler(ConfiguredController<R> controller) {
-    this(
-        new EventDispatcher<>(controller),
+    this(controller.getConfiguration().getConfigurationService().getExecutorService(),
         controller.getConfiguration().getName(),
-        GenericRetry.fromConfiguration(controller.getConfiguration().getRetryConfiguration()),
-        controller.getConfiguration().getConfigurationService().concurrentReconciliationThreads(),
-        controller.getConfiguration().getConfigurationService().getTerminationTimeoutSeconds());
+        new EventDispatcher<>(controller),
+        GenericRetry.fromConfiguration(controller.getConfiguration().getRetryConfiguration()));
   }
 
-  DefaultEventHandler(EventDispatcher<R> dispatcher, String relatedControllerName, Retry retry) {
-    this(
-        dispatcher,
-        relatedControllerName,
-        retry,
-        ConfigurationService.DEFAULT_RECONCILIATION_THREADS_NUMBER,
-        ConfigurationService.DEFAULT_TERMINATION_TIMEOUT_SECONDS);
+  DefaultEventHandler(EventDispatcher<R> eventDispatcher, String relatedControllerName,
+      Retry retry) {
+    this(null, relatedControllerName, eventDispatcher, retry);
   }
 
-  private DefaultEventHandler(
-      EventDispatcher<R> eventDispatcher,
-      String relatedControllerName,
-      Retry retry,
-      int concurrentReconciliationThreads,
-      int terminationTimeout) {
+  private DefaultEventHandler(ExecutorService executor, String relatedControllerName,
+      EventDispatcher<R> eventDispatcher, Retry retry) {
+    this.executor =
+        executor == null
+            ? new ScheduledThreadPoolExecutor(
+                ConfigurationService.DEFAULT_RECONCILIATION_THREADS_NUMBER)
+            : executor;
+    this.controllerName = relatedControllerName;
     this.eventDispatcher = eventDispatcher;
     this.retry = retry;
-    this.controllerName = relatedControllerName;
     eventBuffer = new EventBuffer();
-    this.terminationTimeout = terminationTimeout;
-    executor =
-        new ScheduledThreadPoolExecutor(
-            concurrentReconciliationThreads,
-            runnable -> new Thread(runnable, "EventHandler-" + relatedControllerName));
-  }
-
-  @Override
-  public void close() {
-    try {
-      log.debug("Closing handler for {}", controllerName);
-      executor.shutdown();
-      if (!executor.awaitTermination(terminationTimeout, TimeUnit.SECONDS)) {
-        executor.shutdownNow(); // if we timed out, waiting, cancel everything
-      }
-    } catch (InterruptedException e) {
-      log.debug("Exception closing handler for {}: {}", controllerName, e.getLocalizedMessage());
-    }
   }
 
   public void setEventSourceManager(DefaultEventSourceManager<R> eventSourceManager) {
@@ -146,7 +123,13 @@ private void executeBufferedEvents(String customResourceUid) {
               latestCustomResource.get(),
               retryInfo(customResourceUid));
       log.debug("Executing events for custom resource. Scope: {}", executionScope);
-      executor.execute(new ExecutionConsumer(executionScope, eventDispatcher, this));
+      executor.execute(() -> {
+        // change thread name for easier debugging
+        Thread.currentThread().setName("EventHandler-" + controllerName);
+        PostExecutionControl<R> postExecutionControl =
+            eventDispatcher.handleExecution(executionScope);
+        eventProcessingFinished(executionScope, postExecutionControl);
+      });
     } else {
       log.debug(
           "Skipping executing controller for resource id: {}. Events in queue: {}."

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/ExecutionConsumer.java

@@ -51,10 +51,6 @@ public DefaultEventSourceManager(ConfiguredController<R> controller) {
         new CustomResourceEventSource<>(controller));
   }
 
-  public DefaultEventHandler<R> getEventHandler() {
-    return defaultEventHandler;
-  }
-
   @Override
   public void close() {
     try {