unit tests passing

Author: csviri

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

@@ -236,7 +236,7 @@ public ResourceClassResolver getResourceClassResolver() {
   }
 
   /**
-   * @deprecated Use {@link ConfigurationServiceProvider#overrideCurrent(Consumer)} instead
+   * @deprecated Use {@link ConfigurationService#overrideCurrent(Consumer)} instead
    * @param original that will be overriding
    * @return current overrider
    */

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

@@ -30,12 +30,13 @@ public ExecutorServiceManager(ConfigurationService configurationService) {
     this.configurationService = configurationService;
   }
 
-  public void init() {
+  public ExecutorServiceManager init() {
     final var executorService = configurationService.getExecutorService();
     final var workflowExecutorService = configurationService.getWorkflowExecutorService();
     this.cachingExecutorService = Executors.newCachedThreadPool();
     this.executor = new InstrumentedExecutorService(executorService);
     this.workflowExecutor = new InstrumentedExecutorService(workflowExecutorService);
+    return this;
   }
 
   /**

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

@@ -2,6 +2,7 @@
 
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.ExecutorService;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.KubernetesClient;
@@ -32,4 +33,9 @@ <R> Optional<R> getSecondaryResource(Class<R> expectedType,
   EventSourceRetriever<P> eventSourceRetriever();
 
   KubernetesClient getClient();
+
+  /**
+   * ExecutorService initialized by framework for workflows. Used for workflow standalone mode.
+   */
+  ExecutorService getWorkflowExecutorService();
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/Context.java

@@ -2,6 +2,7 @@
 
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.ExecutorService;
 import java.util.stream.Collectors;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
@@ -75,6 +76,13 @@ public KubernetesClient getClient() {
     return controller.getClient();
   }
 
+  @Override
+  public ExecutorService getWorkflowExecutorService() {
+    // not that this should be always received from executor service manager, so we are able to do
+    // restarts.
+    return controller.getExecutorServiceManager().executorService();
+  }
+
   public DefaultContext<P> setRetryInfo(RetryInfo retryInfo) {
     this.retryInfo = retryInfo;
     return this;

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContext.java

@@ -64,6 +64,7 @@ public class Controller<P extends HasMetadata>
   private final EventProcessor<P> eventProcessor;
   private final ControllerHealthInfo controllerHealthInfo;
   private final ConfigurationService configurationService;
+  private final ExecutorServiceManager executorServiceManager;
 
   public Controller(Reconciler<P> reconciler,
       ControllerConfiguration<P> configuration,
@@ -76,11 +77,13 @@ public Controller(Reconciler<P> reconciler,
     this.configurationService = configuration.getConfigurationService();
     this.kubernetesClient = kubernetesClient;
     this.metrics = Optional.ofNullable(configurationService.getMetrics()).orElse(Metrics.NOOP);
+    this.executorServiceManager = executorServiceManager;
     contextInitializer = reconciler instanceof ContextInitializer;
     isCleaner = reconciler instanceof Cleaner;
 
     final var managed = configurationService.getWorkflowFactory().workflowFor(configuration);
-    managedWorkflow = managed.resolve(kubernetesClient, configuration);
+    managedWorkflow = managed.resolve(kubernetesClient, configuration,
+        executorServiceManager.workflowExecutorService());
 
     eventSourceManager = new EventSourceManager<>(this, executorServiceManager);
     eventProcessor =
@@ -425,4 +428,8 @@ public GroupVersionKind getAssociatedGroupVersionKind() {
   public EventProcessor<P> getEventProcessor() {
     return eventProcessor;
   }
+
+  public ExecutorServiceManager getExecutorServiceManager() {
+    return executorServiceManager;
+  }
 }

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

@@ -4,6 +4,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.ExecutorService;
 import java.util.stream.Collectors;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
@@ -78,7 +79,7 @@ public boolean isEmpty() {
   @Override
   @SuppressWarnings("unchecked")
   public Workflow<P> resolve(KubernetesClient client,
-      ControllerConfiguration<P> configuration) {
+      ControllerConfiguration<P> configuration, ExecutorService executorService) {
     final var alreadyResolved = new HashMap<String, DependentResourceNode>(orderedSpecs.size());
     for (DependentResourceSpec spec : orderedSpecs) {
       final var node = new DependentResourceNode(spec.getName(),

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

@@ -8,8 +8,6 @@
 import java.util.Set;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.javaoperatorsdk.operator.api.config.BaseConfigurationService;
-import io.javaoperatorsdk.operator.api.config.ExecutorServiceManager;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.Deleter;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
@@ -58,7 +56,6 @@ protected DefaultWorkflow(Map<String, DependentResourceNode> dependentResourceNo
       boolean hasCleaner) {
     this.throwExceptionAutomatically = throwExceptionAutomatically;
     this.hasCleaner = hasCleaner;
-
     this.topLevelResources = topLevelResources;
     this.bottomLevelResource = bottomLevelResource;
     this.dependentResourceNodes = dependentResourceNodes;
@@ -88,8 +85,7 @@ private Map<String, DependentResourceNode> toMap(Set<DependentResourceNode> node
   @Override
   public WorkflowReconcileResult reconcile(P primary, Context<P> context) {
     WorkflowReconcileExecutor<P> workflowReconcileExecutor =
-        new WorkflowReconcileExecutor<>(this, primary, context,
-            new ExecutorServiceManager(new BaseConfigurationService()));
+        new WorkflowReconcileExecutor<>(this, primary, context);
     var result = workflowReconcileExecutor.reconcile();
     if (throwExceptionAutomatically) {
       result.throwAggregateExceptionIfErrorsPresent();
@@ -100,8 +96,7 @@ public WorkflowReconcileResult reconcile(P primary, Context<P> context) {
   @Override
   public WorkflowCleanupResult cleanup(P primary, Context<P> context) {
     WorkflowCleanupExecutor<P> workflowCleanupExecutor =
-        new WorkflowCleanupExecutor<>(this, primary, context,
-            new ExecutorServiceManager(new BaseConfigurationService()));
+        new WorkflowCleanupExecutor<>(this, primary, context);
     var result = workflowCleanupExecutor.cleanup();
     if (throwExceptionAutomatically) {
       result.throwAggregateExceptionIfErrorsPresent();

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

@@ -2,6 +2,7 @@
 
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.ExecutorService;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.KubernetesClient;
@@ -23,5 +24,6 @@ default boolean isEmpty() {
     return true;
   }
 
-  Workflow<P> resolve(KubernetesClient client, ControllerConfiguration<P> configuration);
+  Workflow<P> resolve(KubernetesClient client, ControllerConfiguration<P> configuration,
+      ExecutorService executorService);
 }

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

@@ -1,5 +1,7 @@
 package io.javaoperatorsdk.operator.processing.dependent.workflow;
 
+import java.util.concurrent.ExecutorService;
+
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 
@@ -21,7 +23,8 @@ public boolean isEmpty() {
         }
 
         @Override
-        public Workflow resolve(KubernetesClient client, ControllerConfiguration configuration) {
+        public Workflow resolve(KubernetesClient client, ControllerConfiguration configuration,
+            ExecutorService executorService) {
           return new DefaultWorkflow(null);
         }
       };

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

@@ -3,14 +3,14 @@
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.javaoperatorsdk.operator.api.config.ExecutorServiceManager;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.Deleter;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
@@ -23,12 +23,11 @@ public class WorkflowCleanupExecutor<P extends HasMetadata> extends AbstractWork
   private final Set<DependentResourceNode> postDeleteConditionNotMet =
       ConcurrentHashMap.newKeySet();
   private final Set<DependentResourceNode> deleteCalled = ConcurrentHashMap.newKeySet();
-  private final ExecutorServiceManager executorServiceManager;
+  private final ExecutorService executorService;
 
-  public WorkflowCleanupExecutor(Workflow<P> workflow, P primary, Context<P> context,
-      ExecutorServiceManager executorServiceManager) {
+  public WorkflowCleanupExecutor(Workflow<P> workflow, P primary, Context<P> context) {
     super(workflow, primary, context);
-    this.executorServiceManager = executorServiceManager;
+    this.executorService = context.getWorkflowExecutorService();
   }
 
   public synchronized WorkflowCleanupResult cleanup() {
@@ -56,7 +55,7 @@ private synchronized void handleCleanup(DependentResourceNode dependentResourceN
       return;
     }
 
-    Future<?> nodeFuture = executorServiceManager.workflowExecutorService()
+    Future<?> nodeFuture = executorService
         .submit(new CleanupExecutor<>(dependentResourceNode));
     markAsExecuting(dependentResourceNode, nodeFuture);
     log.debug("Submitted for cleanup: {}", dependentResourceNode);

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

@@ -4,14 +4,14 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.javaoperatorsdk.operator.api.config.ExecutorServiceManager;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.Deleter;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
@@ -33,12 +33,11 @@ public class WorkflowReconcileExecutor<P extends HasMetadata> extends AbstractWo
   private final Set<DependentResourceNode> reconciled = ConcurrentHashMap.newKeySet();
   private final Map<DependentResource, ReconcileResult> reconcileResults =
       new ConcurrentHashMap<>();
-  private final ExecutorServiceManager executorServiceManager;
+  private final ExecutorService executorService;
 
-  public WorkflowReconcileExecutor(Workflow<P> workflow, P primary, Context<P> context,
-      ExecutorServiceManager executorServiceManager) {
+  public WorkflowReconcileExecutor(Workflow<P> workflow, P primary, Context<P> context) {
     super(workflow, primary, context);
-    this.executorServiceManager = executorServiceManager;
+    this.executorService = context.getWorkflowExecutorService();
   }
 
   public synchronized WorkflowReconcileResult reconcile() {
@@ -71,7 +70,7 @@ private synchronized <R> void handleReconcile(DependentResourceNode<R, P> depend
     if (!reconcileConditionMet) {
       handleReconcileConditionNotMet(dependentResourceNode);
     } else {
-      Future<?> nodeFuture = executorServiceManager.workflowExecutorService()
+      Future<?> nodeFuture = executorService
           .submit(new NodeReconcileExecutor(dependentResourceNode));
       markAsExecuting(dependentResourceNode, nodeFuture);
       log.debug("Submitted to reconcile: {}", dependentResourceNode);
@@ -89,7 +88,7 @@ private synchronized void handleDelete(DependentResourceNode dependentResourceNo
       return;
     }
 
-    Future<?> nodeFuture = executorServiceManager.workflowExecutorService()
+    Future<?> nodeFuture = executorService
         .submit(new NodeDeleteExecutor(dependentResourceNode));
     markAsExecuting(dependentResourceNode, nodeFuture);
     log.debug("Submitted to delete: {}", dependentResourceNode);

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

@@ -147,6 +147,7 @@ public final void registerEventSource(EventSource eventSource) throws OperatorEx
     registerEventSource(null, eventSource);
   }
 
+  @SuppressWarnings("rawtypes")
   public final synchronized void registerEventSource(String name, EventSource eventSource)
       throws OperatorException {
     Objects.requireNonNull(eventSource, "EventSource must not be null");
@@ -155,8 +156,10 @@ public final synchronized void registerEventSource(String name, EventSource even
         name = EventSourceInitializer.generateNameFor(eventSource);
       }
       if (eventSource instanceof ManagedInformerEventSource) {
-        ((ManagedInformerEventSource) eventSource).setConfigurationService(controller
+        var managedInformerEventSource = ((ManagedInformerEventSource) eventSource);
+        managedInformerEventSource.setConfigurationService(controller
             .getConfiguration().getConfigurationService());
+        managedInformerEventSource.setExecutorServiceManager(executorServiceManager);
       }
       final var named = new NamedEventSource(eventSource, name);
       eventSources.add(named);

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

@@ -34,6 +34,7 @@ void initControllerEventSource(Controller<R> controller) {
     controllerResourceEventSource = new ControllerResourceEventSource<>(controller);
     controllerResourceEventSource
         .setConfigurationService(controller.getConfiguration().getConfigurationService());
+    controllerResourceEventSource.setExecutorServiceManager(controller.getExecutorServiceManager());
   }
 
   ControllerResourceEventSource<R> controllerResourceEventSource() {

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

@@ -17,6 +17,7 @@
 import io.fabric8.kubernetes.client.dsl.Resource;
 import io.fabric8.kubernetes.client.informers.ResourceEventHandler;
 import io.javaoperatorsdk.operator.api.config.ConfigurationService;
+import io.javaoperatorsdk.operator.api.config.ExecutorServiceManager;
 import io.javaoperatorsdk.operator.api.config.NamespaceChangeable;
 import io.javaoperatorsdk.operator.api.config.ResourceConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.RecentOperationCacheFiller;
@@ -179,4 +180,8 @@ public void setConfigurationService(ConfigurationService configurationService) {
     this.configurationService = configurationService;
   }
 
+  public void setExecutorServiceManager(
+      ExecutorServiceManager executorServiceManager) {
+    cache.setExecutorServiceManager(executorServiceManager);
+  }
 }