perf(engine): make I/O executor model configurable, default to virtua…

maestro-engine/src/main/java/com/netflix/maestro/engine/concurrency/EngineExecutors.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2024 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package com.netflix.maestro.engine.concurrency;
+
+import com.netflix.maestro.engine.properties.ThreadingModel;
+import com.netflix.maestro.engine.properties.ThreadingProperties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicLong;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Factory for engine I/O executors.
+ *
+ * <p>Selects between Java 21 virtual threads and a bounded platform-thread pool based on {@link
+ * ThreadingProperties}. Virtual threads lift the previous hard cap on concurrent in-flight JDBC and
+ * signal lookups while preserving a one-line rollback path.
+ */
+@Slf4j
+public final class EngineExecutors {
+  private EngineExecutors() {}
+
+  /**
+   * Build a fresh I/O-bound executor based on the supplied threading properties.
+   *
+   * <p>Callers own the returned executor's lifecycle; call {@link ExecutorService#shutdown()} on
+   * application stop.
+   *
+   * @param props the threading properties; must not be null
+   * @return a new executor service
+   */
+  public static ExecutorService newIoExecutor(ThreadingProperties props) {
+    ThreadingModel model = props.getModel() == null ? ThreadingModel.VIRTUAL : props.getModel();
+    if (model == ThreadingModel.VIRTUAL) {
+      String prefix =
+          props.getVirtualThreadNamePrefix() == null
+              ? "maestro-vt-"
+              : props.getVirtualThreadNamePrefix();
+      LOG.info("Creating virtual-thread I/O executor (name-prefix={})", prefix);
+      return Executors.newThreadPerTaskExecutor(virtualThreadFactory(prefix));
+    }
+    int size = Math.max(1, props.getPlatformThreadPoolSize());
+    LOG.info("Creating platform-thread I/O executor (size={})", size);
+    return Executors.newFixedThreadPool(size, daemonThreadFactory("maestro-io-"));
+  }
+
+  private static ThreadFactory virtualThreadFactory(String prefix) {
+    AtomicLong seq = new AtomicLong();
+    return runnable -> {
+      Thread t = Thread.ofVirtual().name(prefix + seq.incrementAndGet()).unstarted(runnable);
+      return t;
+    };
+  }
+
+  private static ThreadFactory daemonThreadFactory(String prefix) {
+    AtomicLong seq = new AtomicLong();
+    return runnable -> {
+      Thread t = new Thread(runnable, prefix + seq.incrementAndGet());
+      t.setDaemon(true);
+      return t;
+    };
+  }
+}

maestro-engine/src/main/java/com/netflix/maestro/engine/eval/MaestroParamExtensionRepo.java

@@ -15,34 +15,59 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.netflix.maestro.annotations.Nullable;
+import com.netflix.maestro.engine.concurrency.EngineExecutors;
 import com.netflix.maestro.engine.dao.MaestroStepInstanceDao;
 import com.netflix.maestro.engine.handlers.SignalHandler;
+import com.netflix.maestro.engine.properties.ThreadingProperties;
 import com.netflix.maestro.exceptions.MaestroUnprocessableEntityException;
 import com.netflix.maestro.utils.Checks;
 import com.netflix.sel.ext.Extension;
 import com.netflix.sel.type.SelUtilFunc;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
 
-/** A repository to hold maestro param extensions for the param evaluation. */
+/**
+ * A repository to hold maestro param extensions for the param evaluation.
+ *
+ * <p>Owns the I/O-bound executor that backs {@link MaestroParamExtension}'s blocking JDBC and
+ * signal lookups. The executor is constructed by {@link EngineExecutors} during {@link
+ * #initialize()} according to the supplied {@link ThreadingProperties}, and is shut down on
+ * {@link #shutdown()}.
+ */
 @Slf4j
 public class MaestroParamExtensionRepo {
-  private static final int THREAD_NUM = 3;
   private final ThreadLocal<Extension> repos = new ThreadLocal<>();
   private final MaestroStepInstanceDao stepInstanceDao;
   private final ObjectMapper objectMapper;
   private final String env;
-  private ExecutorService executor;
+  private final ThreadingProperties threadingProperties;
+  private ExecutorService ioExecutor;
 
-  /** Constructor. */
+  /** Constructor with default threading properties (virtual threads). */
   public MaestroParamExtensionRepo(
       MaestroStepInstanceDao stepInstanceDao, String env, ObjectMapper objectMapper) {
+    this(stepInstanceDao, env, objectMapper, new ThreadingProperties());
+  }
+
+  /**
+   * Constructor.
+   *
+   * @param stepInstanceDao DAO used by the param extension for JDBC lookups
+   * @param env execution environment identifier
+   * @param objectMapper Jackson object mapper used to deserialize runtime summaries
+   * @param threadingProperties controls the executor model (virtual vs. platform) and tuning
+   */
+  public MaestroParamExtensionRepo(
+      MaestroStepInstanceDao stepInstanceDao,
+      String env,
+      ObjectMapper objectMapper,
+      ThreadingProperties threadingProperties) {
     this.stepInstanceDao = stepInstanceDao;
     this.objectMapper = objectMapper;
     this.env = env;
+    this.threadingProperties = threadingProperties;
   }
 
   /** Reset repo by creating a new param extension wrapper for the current thread. */
@@ -52,7 +77,7 @@ public void reset(
       InstanceWrapper instanceWrapper) {
     Extension ext =
         new MaestroParamExtension(
-            executor,
+            ioExecutor,
             stepInstanceDao,
             env,
             allStepOutputData,
@@ -72,20 +97,30 @@ public Extension get() {
     return repos.get();
   }
 
-  /** Initialize the ExtensionRepo. */
+  /** Initialize the ExtensionRepo and its I/O executor. */
   void initialize() {
     LOG.info("Initializing ExtensionRepo within Spring boot...");
     SelUtilFunc.register("toJson", this::toJsonExtFunction);
-    executor = Executors.newFixedThreadPool(THREAD_NUM);
-    ((ThreadPoolExecutor) executor).prestartAllCoreThreads();
+    ioExecutor = EngineExecutors.newIoExecutor(threadingProperties);
   }
 
-  /** Gracefully shutdown the ExtensionRepo. */
-  @SuppressWarnings({"PMD.NullAssignment"})
+  /** Gracefully shutdown the ExtensionRepo and its I/O executor. */
   void shutdown() {
     LOG.info("Shutdown ExtensionRepo within Spring boot...");
-    executor.shutdown();
-    executor = null;
+    ExecutorService toShutdown = ioExecutor;
+    ioExecutor = null;
+    if (toShutdown == null) {
+      return;
+    }
+    toShutdown.shutdown();
+    try {
+      if (!toShutdown.awaitTermination(5, TimeUnit.SECONDS)) {
+        toShutdown.shutdownNow();
+      }
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      toShutdown.shutdownNow();
+    }
   }
 
   // Add a SEL function to convert the input object to a JSON string. If there are more, will

maestro-engine/src/main/java/com/netflix/maestro/engine/properties/ThreadingModel.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2024 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package com.netflix.maestro.engine.properties;
+
+/**
+ * Threading model selection for engine I/O executors.
+ *
+ * <p>Used to gate a safe, incremental rollout of Java 21 virtual threads. {@link #VIRTUAL} is the
+ * recommended default; {@link #PLATFORM} is preserved as a rollback target for environments where
+ * virtual threads are not yet desirable.
+ */
+public enum ThreadingModel {
+  /** Java 21 virtual threads via {@link java.util.concurrent.Executors#newThreadPerTaskExecutor}. */
+  VIRTUAL,
+
+  /**
+   * Legacy bounded platform-thread pool, retained only for safe rollback. Hard-caps the number of
+   * concurrent in-flight I/O calls and is therefore a throughput ceiling.
+   */
+  PLATFORM
+}

maestro-engine/src/main/java/com/netflix/maestro/engine/properties/ThreadingProperties.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2024 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package com.netflix.maestro.engine.properties;
+
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * Threading model configuration for engine I/O executors.
+ *
+ * <p>Bound from {@code maestro.threading.*} in {@code application.yml}. Defaults to virtual threads;
+ * flip to {@link ThreadingModel#PLATFORM} for safe rollback if a virtual-thread regression is
+ * observed.
+ */
+@Getter
+@Setter
+public class ThreadingProperties {
+  /** Active threading model. Defaults to virtual threads. */
+  private ThreadingModel model = ThreadingModel.VIRTUAL;
+
+  /** Size of the platform-thread fallback pool. Ignored when {@link #model} is {@code VIRTUAL}. */
+  private int platformThreadPoolSize = 3;
+
+  /** Name prefix for virtual threads spawned by the engine. Useful for observability. */
+  private String virtualThreadNamePrefix = "maestro-vt-";
+}

maestro-engine/src/test/java/com/netflix/maestro/engine/concurrency/EngineExecutorsTest.java

@@ -0,0 +1,107 @@
+/*
+ * Copyright 2024 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package com.netflix.maestro.engine.concurrency;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.netflix.maestro.engine.properties.ThreadingModel;
+import com.netflix.maestro.engine.properties.ThreadingProperties;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.junit.Test;
+
+public class EngineExecutorsTest {
+
+  private static final int CONCURRENT_TASKS = 16;
+  private static final long PER_TASK_SLEEP_MILLIS = 100;
+
+  @Test
+  public void virtualThreadsRunIOTasksConcurrently() throws Exception {
+    ThreadingProperties props = new ThreadingProperties();
+    props.setModel(ThreadingModel.VIRTUAL);
+
+    ExecutorService executor = EngineExecutors.newIoExecutor(props);
+    try {
+      long elapsed = runConcurrentSleepTasks(executor);
+      // With virtual threads, all 16 tasks should run concurrently. The total
+      // wall-clock time should be much less than the serialized 16 * 100 ms = 1600 ms
+      // lower bound. A generous bound accommodates JIT and scheduling jitter.
+      assertThat(elapsed)
+          .as("virtual threads should not serialize independent I/O-bound tasks")
+          .isLessThan(CONCURRENT_TASKS * PER_TASK_SLEEP_MILLIS / 2L);
+    } finally {
+      executor.shutdownNow();
+    }
+  }
+
+  @Test
+  public void platformThreadPoolCapsConcurrencyAtConfiguredSize() throws Exception {
+    ThreadingProperties props = new ThreadingProperties();
+    props.setModel(ThreadingModel.PLATFORM);
+    props.setPlatformThreadPoolSize(2);
+
+    ExecutorService executor = EngineExecutors.newIoExecutor(props);
+    try {
+      long elapsed = runConcurrentSleepTasks(executor);
+      // With a 2-thread pool and 16 tasks each sleeping 100 ms, the total wall-clock
+      // time must be at least 16/2 * 100 = 800 ms. We allow some headroom.
+      long lowerBound =
+          ((long) Math.ceil((double) CONCURRENT_TASKS / props.getPlatformThreadPoolSize()))
+              * PER_TASK_SLEEP_MILLIS;
+      assertThat(elapsed)
+          .as("bounded platform-thread pool must serialize tasks beyond pool size")
+          .isGreaterThanOrEqualTo(lowerBound - 50L);
+    } finally {
+      executor.shutdownNow();
+    }
+  }
+
+  @Test
+  public void virtualThreadExecutorUsesVirtualThreads() throws Exception {
+    ThreadingProperties props = new ThreadingProperties();
+    props.setModel(ThreadingModel.VIRTUAL);
+
+    ExecutorService executor = EngineExecutors.newIoExecutor(props);
+    try {
+      Future<Boolean> isVirtual = executor.submit(() -> Thread.currentThread().isVirtual());
+      assertThat(isVirtual.get(1, TimeUnit.SECONDS)).isTrue();
+    } finally {
+      executor.shutdownNow();
+    }
+  }
+
+  private static long runConcurrentSleepTasks(ExecutorService executor)
+      throws InterruptedException, java.util.concurrent.ExecutionException, java.util.concurrent.TimeoutException {
+    CountDownLatch start = new CountDownLatch(1);
+    List<Future<Long>> futures = new ArrayList<>();
+    long startNanos = System.nanoTime();
+    for (int i = 0; i < CONCURRENT_TASKS; i++) {
+      futures.add(
+          executor.submit(
+              () -> {
+                start.await();
+                Thread.sleep(PER_TASK_SLEEP_MILLIS);
+                return Thread.currentThread().threadId();
+              }));
+    }
+    start.countDown();
+    for (Future<Long> f : futures) {
+      f.get(5, TimeUnit.SECONDS);
+    }
+    return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+  }
+}

maestro-server/src/main/java/com/netflix/maestro/server/config/MaestroEngineConfiguration.java

@@ -130,10 +130,14 @@ public ParamsManager paramsManager(DefaultParamManager defaultParamManager) {
   public MaestroParamExtensionRepo maestroParamExtensionRepo(
       MaestroStepInstanceDao stepInstanceDao,
       StepRuntimeProperties stepRuntimeProperties,
+      MaestroProperties maestroProperties,
       @Qualifier(Constants.MAESTRO_QUALIFIER) ObjectMapper objectMapper) {
     LOG.info("Creating Maestro MaestroParamExtensionRepo within Spring boot...");
     return new MaestroParamExtensionRepo(
-        stepInstanceDao, stepRuntimeProperties.getEnv(), objectMapper);
+        stepInstanceDao,
+        stepRuntimeProperties.getEnv(),
+        objectMapper,
+        maestroProperties.getThreading());
   }
 
   @Bean(initMethod = "postConstruct", destroyMethod = "preDestroy")