fix: run event source start on specific thread pool

Fixes #1603

Author: metacosm

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

@@ -5,6 +5,7 @@
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -17,6 +18,9 @@ public class ExecutorServiceManager {
   private static ExecutorServiceManager instance;
   private final ExecutorService executor;
   private final ExecutorService workflowExecutor;
+
+  private static final ForkJoinPool threadPool = new ForkJoinPool(
+      Runtime.getRuntime().availableProcessors());
   private final int terminationTimeoutSeconds;
 
   private ExecutorServiceManager(ExecutorService executor, ExecutorService workflowExecutor,
@@ -68,6 +72,21 @@ public ExecutorService workflowExecutorService() {
     return workflowExecutor;
   }
 
+  public static void executeInParallel(Runnable callable) {
+    executeInParallel(() -> {
+      callable.run();
+      return null;
+    });
+  }
+
+  public static <T> T executeInParallel(Callable<T> callable) {
+    try {
+      return threadPool.submit(callable).get();
+    } catch (InterruptedException | ExecutionException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
   private void doStop() {
     try {
       log.debug("Closing executor");
@@ -80,6 +99,7 @@ private void doStop() {
         executor.shutdownNow(); // if we timed out, waiting, cancel everything
       }
 
+      threadPool.shutdown();
     } catch (InterruptedException e) {
       log.debug("Exception closing executor: {}", e.getLocalizedMessage());
     }

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

@@ -13,6 +13,7 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.MissingCRDException;
 import io.javaoperatorsdk.operator.OperatorException;
+import io.javaoperatorsdk.operator.api.config.ExecutorServiceManager;
 import io.javaoperatorsdk.operator.api.config.NamespaceChangeable;
 import io.javaoperatorsdk.operator.api.reconciler.EventSourceInitializer;
 import io.javaoperatorsdk.operator.processing.Controller;
@@ -65,20 +66,24 @@ public void postProcessDefaultEventSourcesAfterProcessorInitializer() {
   @Override
   public synchronized void start() {
     startEventSource(eventSources.namedControllerResourceEventSource());
-    eventSources.additionalNamedEventSources()
+
+    // starting event sources on the workflow executor which shouldn't be used at this point
+    ExecutorServiceManager.executeInParallel(() -> eventSources.additionalNamedEventSources()
         .filter(es -> es.priority().equals(EventSourceStartPriority.RESOURCE_STATE_LOADER))
-        .parallel().forEach(this::startEventSource);
-    eventSources.additionalNamedEventSources()
+        .parallel()
+        .forEach(this::startEventSource));
+
+    ExecutorServiceManager.executeInParallel(() -> eventSources.additionalNamedEventSources()
         .filter(es -> es.priority().equals(EventSourceStartPriority.DEFAULT))
-        .parallel().forEach(this::startEventSource);
+        .parallel().forEach(this::startEventSource));
   }
 
   @Override
   public synchronized void stop() {
     stopEventSource(eventSources.namedControllerResourceEventSource());
-    eventSources.additionalNamedEventSources().parallel().forEach(this::stopEventSource);
+    ExecutorServiceManager.executeInParallel(
+        () -> eventSources.additionalNamedEventSources().parallel().forEach(this::stopEventSource));
     eventSources.clear();
-
   }
 
   @SuppressWarnings("rawtypes")

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManager.java

@@ -24,6 +24,7 @@
 import io.javaoperatorsdk.operator.ReconcilerUtils;
 import io.javaoperatorsdk.operator.api.config.Cloner;
 import io.javaoperatorsdk.operator.api.config.ConfigurationServiceProvider;
+import io.javaoperatorsdk.operator.api.config.ExecutorServiceManager;
 import io.javaoperatorsdk.operator.api.config.ResourceConfiguration;
 import io.javaoperatorsdk.operator.processing.LifecycleAware;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
@@ -45,7 +46,8 @@ public class InformerManager<T extends HasMetadata, C extends ResourceConfigurat
 
   @Override
   public void start() throws OperatorException {
-    sources.values().parallelStream().forEach(LifecycleAware::start);
+    ExecutorServiceManager
+        .executeInParallel(() -> sources.values().parallelStream().forEach(LifecycleAware::start));
   }
 
   void initSources(MixedOperation<T, KubernetesResourceList<T>, Resource<T>> client,
@@ -86,18 +88,19 @@ public void changeNamespaces(Set<String> namespaces) {
     log.debug("Stopped informer {} for namespaces: {}", this, sourcesToRemove);
     sourcesToRemove.forEach(k -> sources.remove(k).stop());
 
-    namespaces.forEach(ns -> {
-      if (!sources.containsKey(ns)) {
-        final var source =
-            createEventSource(
-                client.inNamespace(ns).withLabelSelector(configuration.getLabelSelector()),
-                eventHandler, ns);
-        source.addIndexers(this.indexers);
-        source.start();
-        log.debug("Registered new {} -> {} for namespace: {}", this, source,
-            ns);
-      }
-    });
+    ExecutorServiceManager.executeInParallel(
+        () -> namespaces.forEach(ns -> {
+          if (!sources.containsKey(ns)) {
+            final var source =
+                createEventSource(
+                    client.inNamespace(ns).withLabelSelector(configuration.getLabelSelector()),
+                    eventHandler, ns);
+            source.addIndexers(this.indexers);
+            source.start();
+            log.debug("Registered new {} -> {} for namespace: {}", this, source,
+                ns);
+          }
+        }));
   }
 
 
@@ -113,15 +116,18 @@ private InformerWrapper<T> createEventSource(
 
   @Override
   public void stop() {
-    log.info("Stopping {}", this);
-    sources.forEach((ns, source) -> {
-      try {
-        log.debug("Stopping informer for namespace: {} -> {}", ns, source);
-        source.stop();
-      } catch (Exception e) {
-        log.warn("Error stopping informer for namespace: {} -> {}", ns, source, e);
-      }
-    });
+    ExecutorServiceManager.executeInParallel(
+        () -> {
+          log.info("Stopping {}", this);
+          sources.forEach((ns, source) -> {
+            try {
+              log.debug("Stopping informer for namespace: {} -> {}", ns, source);
+              source.stop();
+            } catch (Exception e) {
+              log.warn("Error stopping informer for namespace: {} -> {}", ns, source, e);
+            }
+          });
+        });
   }
 
   @Override