simple impl

Author: csviri

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

@@ -80,6 +80,12 @@ default int concurrentReconciliationThreads() {
     return DEFAULT_RECONCILIATION_THREADS_NUMBER;
   }
 
+  int DEFAULT_WORKFLOW_EXECUTOR_THREAD_NUMBER = DEFAULT_RECONCILIATION_THREADS_NUMBER;
+
+  default int concurrentWorkflowExecutorThreads() {
+    return DEFAULT_WORKFLOW_EXECUTOR_THREAD_NUMBER;
+  }
+
   /**
    * Used to clone custom resources. It is strongly suggested that implementors override this method
    * since the default implementation creates a new {@link Cloner} instance each time this method is
@@ -122,6 +128,10 @@ default ExecutorService getExecutorService() {
     return Executors.newFixedThreadPool(concurrentReconciliationThreads());
   }
 
+  default ExecutorService getWorkflowExecutorService() {
+    return Executors.newFixedThreadPool(concurrentWorkflowExecutorThreads());
+  }
+
   default boolean closeClientOnStop() {
     return true;
   }

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

@@ -21,7 +21,8 @@ public class ConfigurationServiceOverrider {
   private int timeoutSeconds;
   private boolean closeClientOnStop;
   private ObjectMapper objectMapper;
-  private ExecutorService executorService = null;
+  private ExecutorService executorService;
+  private ExecutorService workflowExecutorService;
   private LeaderElectionConfiguration leaderElectionConfiguration;
 
   ConfigurationServiceOverrider(ConfigurationService original) {
@@ -77,6 +78,12 @@ public ConfigurationServiceOverrider withExecutorService(ExecutorService executo
     return this;
   }
 
+  public ConfigurationServiceOverrider withWorkflowExecutorService(
+      ExecutorService workflowExecutorService) {
+    this.workflowExecutorService = workflowExecutorService;
+    return this;
+  }
+
   public ConfigurationServiceOverrider withObjectMapper(ObjectMapper objectMapper) {
     this.objectMapper = objectMapper;
     return this;
@@ -134,6 +141,15 @@ public ExecutorService getExecutorService() {
         }
       }
 
+      @Override
+      public ExecutorService getWorkflowExecutorService() {
+        if (workflowExecutorService != null) {
+          return workflowExecutorService;
+        } else {
+          return super.getWorkflowExecutorService();
+        }
+      }
+
       @Override
       public ObjectMapper getObjectMapper() {
         return objectMapper;

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

@@ -17,6 +17,7 @@ public class ExecutorServiceManager {
   private static ExecutorServiceManager instance;
 
   private final ExecutorService executor;
+  private ExecutorService workflowExecutor;
   private final int terminationTimeoutSeconds;
 
   private ExecutorServiceManager(InstrumentedExecutorService executor,
@@ -60,13 +61,28 @@ public ExecutorService executorService() {
     return executor;
   }
 
+  // needs to be synchronized since only called when a workflow is initialized, but that happens in
+  // controllers which are started in parallel
+  public synchronized ExecutorService workflowExecutorService() {
+    if (workflowExecutor == null) {
+      workflowExecutor =
+          new InstrumentedExecutorService(
+              ConfigurationServiceProvider.instance().getWorkflowExecutorService());
+    }
+    return workflowExecutor;
+  }
+
   private void doStop() {
     try {
       log.debug("Closing executor");
       executor.shutdown();
+      workflowExecutor.shutdown();
       if (!executor.awaitTermination(terminationTimeoutSeconds, TimeUnit.SECONDS)) {
         executor.shutdownNow(); // if we timed out, waiting, cancel everything
       }
+      if (!workflowExecutor.awaitTermination(terminationTimeoutSeconds, TimeUnit.SECONDS)) {
+        executor.shutdownNow(); // if we timed out, waiting, cancel everything
+      }
     } catch (InterruptedException e) {
       log.debug("Exception closing executor: {}", e.getLocalizedMessage());
     }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/Workflow.java

@@ -7,7 +7,7 @@
 import java.util.stream.Collectors;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.javaoperatorsdk.operator.api.config.ConfigurationServiceProvider;
+import io.javaoperatorsdk.operator.api.config.ExecutorServiceManager;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
 
@@ -30,7 +30,7 @@ public class Workflow<P extends HasMetadata> {
   private ExecutorService executorService;
 
   public Workflow(Set<DependentResourceNode> dependentResourceNodes) {
-    this.executorService = ConfigurationServiceProvider.instance().getExecutorService();
+    this.executorService = ExecutorServiceManager.instance().workflowExecutorService();
     this.dependentResourceNodes = dependentResourceNodes;
     this.throwExceptionAutomatically = THROW_EXCEPTION_AUTOMATICALLY_DEFAULT;
     preprocessForReconcile();

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/builder/WorkflowBuilder.java

@@ -76,7 +76,7 @@ public WorkflowBuilder<P> withThrowExceptionFurther(boolean throwExceptionFurthe
 
   public Workflow<P> build() {
     return new Workflow(
-        dependentResourceNodes, ExecutorServiceManager.instance().executorService(),
+        dependentResourceNodes, ExecutorServiceManager.instance().workflowExecutorService(),
         throwExceptionAutomatically);
   }