fix: make sure we run each task on a different thread

Author: metacosm

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

@@ -9,12 +9,13 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import io.javaoperatorsdk.operator.OperatorException;
-
 public class ExecutorServiceManager {
   private static final Logger log = LoggerFactory.getLogger(ExecutorServiceManager.class);
   private static ExecutorServiceManager instance;
@@ -63,6 +64,31 @@ public synchronized static ExecutorServiceManager instance() {
     return instance;
   }
 
+  public static <T> void executeAndWaitForAllToComplete(Stream<T> stream,
+      Function<T, Void> task, Function<T, String> threadNamer) {
+    final var instrumented = new InstrumentedExecutorService(
+        Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()));
+
+    try {
+      instrumented.invokeAll(stream.parallel().map(item -> (Callable<Void>) () -> {
+        // change thread name for easier debugging
+        final var thread = Thread.currentThread();
+        final var name = thread.getName();
+        thread.setName(threadNamer.apply(item));
+        try {
+          task.apply(item);
+          return null;
+        } finally {
+          // restore original name
+          thread.setName(name);
+        }
+      }).collect(Collectors.toList()));
+      shutdown(instrumented);
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+    }
+  }
+
   public ExecutorService executorService() {
     return executor;
   }
@@ -71,36 +97,6 @@ public ExecutorService workflowExecutorService() {
     return workflowExecutor;
   }
 
-  /**
-   * Runs the specified I/O-bound task and waits for its completion using the new ExecutorService
-   *
-   * @param task task to run concurrently
-   * @param threadNamePrefix the prefix with which to prefix thread names when tasks are run this
-   *        way
-   */
-  public static void executeAndWaitForCompletion(Runnable task, String threadNamePrefix) {
-    final var executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
-    ExecutorService instrumented = new InstrumentedExecutorService(executor);
-    // change thread name for easier debugging
-    final var thread = Thread.currentThread();
-    final var name = thread.getName();
-    try {
-      thread.setName(threadNamePrefix + "-" + thread.getId());
-      instrumented.submit(task)
-          .get(ConfigurationServiceProvider.instance().cacheSyncTimeout().toSeconds(),
-              TimeUnit.SECONDS);
-      shutdown(instrumented);
-    } catch (InterruptedException e) {
-      thread.interrupt();
-      throw new OperatorException("Couldn't execute task", e);
-    } catch (ExecutionException | TimeoutException e) {
-      throw new OperatorException("Couldn't execute task", e);
-    } finally {
-      // restore original name
-      thread.setName(name);
-    }
-  }
-
   private void doStop() {
     try {
       log.debug("Closing executor");

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

@@ -4,6 +4,7 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.Set;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
@@ -66,28 +67,34 @@ public void postProcessDefaultEventSourcesAfterProcessorInitializer() {
   public synchronized void start() {
     startEventSource(eventSources.namedControllerResourceEventSource());
 
-    // starting event sources on the workflow executor which shouldn't be used at this point
-    ExecutorServiceManager.executeAndWaitForCompletion(
-        () -> eventSources.additionalNamedEventSources()
-            .filter(es -> es.priority().equals(EventSourceStartPriority.RESOURCE_STATE_LOADER))
-            .parallel()
-            .forEach(this::startEventSource),
-        "LowLevelEventSourceStart");
+    ExecutorServiceManager.executeAndWaitForAllToComplete(
+        eventSources.additionalNamedEventSources()
+            .filter(es -> es.priority().equals(EventSourceStartPriority.RESOURCE_STATE_LOADER)),
+        this::startEventSource,
+        getThreadNamer("start"));
 
-    ExecutorServiceManager.executeAndWaitForCompletion(
-        () -> eventSources.additionalNamedEventSources()
-            .filter(es -> es.priority().equals(EventSourceStartPriority.DEFAULT))
-            .parallel().forEach(this::startEventSource),
-        "DefaultEventSourceStart");
+    ExecutorServiceManager.executeAndWaitForAllToComplete(
+        eventSources.additionalNamedEventSources()
+            .filter(es -> es.priority().equals(EventSourceStartPriority.DEFAULT)),
+        this::startEventSource,
+        getThreadNamer("start"));
+  }
+
+  private static Function<NamedEventSource, String> getThreadNamer(String stage) {
+    return es -> {
+      final var name = es.name();
+      return es.priority() + " " + stage + " -> "
+          + (es.isNameSet() ? name + " " + es.original().getClass().getSimpleName() : name);
+    };
   }
 
   @Override
   public synchronized void stop() {
     stopEventSource(eventSources.namedControllerResourceEventSource());
-    ExecutorServiceManager.executeAndWaitForCompletion(
-        () -> eventSources.additionalNamedEventSources().parallel()
-            .forEach(this::stopEventSource),
-        "EventSourceStop");
+    ExecutorServiceManager.executeAndWaitForAllToComplete(
+        eventSources.additionalNamedEventSources(),
+        this::stopEventSource,
+        getThreadNamer("stop"));
     eventSources.clear();
   }
 
@@ -104,7 +111,7 @@ private void logEventSourceEvent(NamedEventSource eventSource, String event) {
     }
   }
 
-  private void startEventSource(NamedEventSource eventSource) {
+  private Void startEventSource(NamedEventSource eventSource) {
     try {
       logEventSourceEvent(eventSource, "Starting");
       eventSource.start();
@@ -114,16 +121,18 @@ private void startEventSource(NamedEventSource eventSource) {
     } catch (Exception e) {
       throw new OperatorException("Couldn't start source " + eventSource.name(), e);
     }
+    return null;
   }
 
-  private void stopEventSource(NamedEventSource eventSource) {
+  private Void stopEventSource(NamedEventSource eventSource) {
     try {
       logEventSourceEvent(eventSource, "Stopping");
       eventSource.stop();
       logEventSourceEvent(eventSource, "Stopped");
     } catch (Exception e) {
       log.warn("Error closing {} -> {}", eventSource.name(), e);
     }
+    return null;
   }
 
   public final void registerEventSource(EventSource eventSource) throws OperatorException {

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

@@ -3,17 +3,20 @@
 import java.util.Objects;
 
 import io.javaoperatorsdk.operator.OperatorException;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceInitializer;
 import io.javaoperatorsdk.operator.processing.event.source.EventSource;
 import io.javaoperatorsdk.operator.processing.event.source.EventSourceStartPriority;
 
 class NamedEventSource implements EventSource {
 
   private final EventSource original;
   private final String name;
+  private final boolean nameSet;
 
   NamedEventSource(EventSource original, String name) {
     this.original = original;
     this.name = name;
+    nameSet = !name.equals(EventSourceInitializer.generateNameFor(original));
   }
 
   @Override
@@ -63,4 +66,8 @@ public int hashCode() {
   public EventSourceStartPriority priority() {
     return original.priority();
   }
+
+  public boolean isNameSet() {
+    return nameSet;
+  }
 }

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

@@ -110,7 +110,6 @@ 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);