[WIP] feat: virtual thread support

docs/content/en/docs/documentation/operations/configuration.md

@@ -23,6 +23,34 @@ Operator operator = new Operator( override -> override
         .withLeaderElectionConfiguration(new LeaderElectionConfiguration("bar", "barNS")));
 ```
 
+### Virtual Threads
+
+Reconciliation is mostly about blocking: talking to the Kubernetes API server or to external
+systems. Virtual threads make such blocking calls much cheaper than platform threads, and the
+framework can be switched over to them with a single flag:
+
+```java
+Operator operator = new Operator(override -> override.withUseVirtualThreads(true));
+```
+
+When enabled, reconciliations, dependent resource workflows and the framework's internal
+housekeeping (starting the informers, for example) all run on virtual threads.
+
+Enabling virtual threads does **not** remove the concurrency limits, parallelism is configured
+exactly as before: `withConcurrentReconciliationThreads(int)` still caps how many reconciliations
+run at the same time and `withConcurrentWorkflowExecutorThreads(int)` how many dependent resources
+of a workflow are processed concurrently. Only the threads backing those limits change. Since
+virtual threads are cheap, these limits can usually be raised significantly compared to what is
+reasonable with platform threads.
+
+Two things to keep in mind:
+
+- Virtual threads require Java 21 or later at runtime. When the flag is set on an older JVM, a
+  warning is logged and platform threads are used instead, so the same configuration works on any
+  supported Java version.
+- A custom `ExecutorService` provided through `withExecutorService(...)` or
+  `withWorkflowExecutorService(...)` is always used as is, the flag has no effect on it.
+
 ## Reconciler-Level Configuration
 
 While reconcilers are typically configured using the `@ControllerConfiguration` annotation, you can also override configuration at runtime when registering the reconciler with the operator. You can either:
@@ -265,6 +293,7 @@ All operator-level keys are prefixed with `josdk.`.
 |---|---|---|
 | `josdk.check-crd` | `Boolean` | Validate CRDs against local model on startup |
 | `josdk.close-client-on-stop` | `Boolean` | Close the Kubernetes client when the operator stops |
+| `josdk.use-virtual-threads` | `Boolean` | Run the framework's concurrent work on virtual threads (requires Java 21+ at runtime) |
 | `josdk.use-ssa-to-patch-primary-resource` | `Boolean` | Use Server-Side Apply to patch the primary resource |
 | `josdk.clone-secondary-resources-when-getting-from-cache` | `Boolean` | Clone secondary resources on cache reads |
 

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

@@ -20,7 +20,6 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.function.Consumer;
 
 import org.slf4j.Logger;
@@ -228,6 +227,34 @@ default Metrics getMetrics() {
     return Metrics.NOOP;
   }
 
+  /**
+   * Whether the framework should run the tasks it executes concurrently — reconciliations,
+   * dependent workflows and internal housekeeping such as starting the informers — on virtual
+   * threads instead of platform threads.
+   *
+   * <p>Virtual threads make blocking operations, which is essentially all a reconciler does while
+   * talking to the Kubernetes API server or to external systems, much cheaper. Enabling them does
+   * <em>not</em> lift the configured concurrency limits: {@link #concurrentReconciliationThreads()}
+   * and {@link #concurrentWorkflowExecutorThreads()} still cap how many reconciliations,
+   * respectively dependent resources, are processed at the same time, they just aren't backed by a
+   * pool of platform threads anymore. Since virtual threads are cheap, those limits can be set
+   * considerably higher than what would be reasonable for platform threads.
+   *
+   * <p>Requires Java 21 or later at runtime. When enabled on an older JVM, a warning is logged and
+   * platform threads are used, so that the same configuration works regardless of the Java version
+   * the operator runs on.
+   *
+   * <p>Note that this only affects the executors created by the framework: a custom {@link
+   * ExecutorService} provided through {@link #getExecutorService()} or {@link
+   * #getWorkflowExecutorService()} is used as is.
+   *
+   * @return {@code true} to use virtual threads, {@code false} (default) to use platform threads
+   * @since 5.7.0
+   */
+  default boolean useVirtualThreads() {
+    return false;
+  }
+
   /**
    * Override to provide a custom {@link ExecutorService} implementation to change how threads
    * handle concurrent reconciliations
@@ -236,7 +263,8 @@ default Metrics getMetrics() {
    *     processing
    */
   default ExecutorService getExecutorService() {
-    return Executors.newFixedThreadPool(concurrentReconciliationThreads());
+    return ExecutorServiceManager.newBoundedExecutorService(
+        concurrentReconciliationThreads(), useVirtualThreads());
   }
 
   /**
@@ -246,7 +274,8 @@ default ExecutorService getExecutorService() {
    * @return the {@link ExecutorService} implementation to use for dependent workflow processing
    */
   default ExecutorService getWorkflowExecutorService() {
-    return Executors.newFixedThreadPool(concurrentWorkflowExecutorThreads());
+    return ExecutorServiceManager.newBoundedExecutorService(
+        concurrentWorkflowExecutorThreads(), useVirtualThreads());
   }
 
   /**

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

@@ -50,6 +50,7 @@ public class ConfigurationServiceOverrider {
   private KubernetesClient client;
   private ExecutorService executorService;
   private ExecutorService workflowExecutorService;
+  private Boolean useVirtualThreads;
   private LeaderElectionConfiguration leaderElectionConfiguration;
   private String clusterScopedEventNamespace;
   private EventRecorder eventRecorder;
@@ -119,6 +120,19 @@ public ConfigurationServiceOverrider withWorkflowExecutorService(
     return this;
   }
 
+  /**
+   * Makes the framework run the tasks it executes concurrently on virtual threads instead of
+   * platform threads. Requires Java 21 or later at runtime, see {@link
+   * ConfigurationService#useVirtualThreads()} for the details.
+   *
+   * @param useVirtualThreads {@code true} to use virtual threads
+   * @return this {@link ConfigurationServiceOverrider} for chained customization
+   */
+  public ConfigurationServiceOverrider withUseVirtualThreads(boolean useVirtualThreads) {
+    this.useVirtualThreads = useVirtualThreads;
+    return this;
+  }
+
   /**
    * Replaces the default {@link KubernetesClient} instance by the specified one. This is the
    * preferred mechanism to configure which client will be used to access the cluster.
@@ -322,6 +336,11 @@ public boolean closeClientOnStop() {
         return overriddenValueOrDefault(closeClientOnStop, ConfigurationService::closeClientOnStop);
       }
 
+      @Override
+      public boolean useVirtualThreads() {
+        return overriddenValueOrDefault(useVirtualThreads, ConfigurationService::useVirtualThreads);
+      }
+
       @Override
       public ExecutorService getExecutorService() {
         if (executorService != null) {

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

@@ -49,6 +49,37 @@ public class ExecutorServiceManager {
     start(configurationService);
   }
 
+  /**
+   * Creates the executor service used to run a bounded number of tasks concurrently, either backed
+   * by virtual threads or by a fixed size pool of platform threads. The concurrency limit is
+   * enforced in both cases.
+   *
+   * @param maxConcurrency the maximal number of tasks executed at the same time
+   * @param useVirtualThreads whether virtual threads should be used, see {@link
+   *     ConfigurationService#useVirtualThreads()}
+   * @return the created {@link ExecutorService}
+   */
+  public static ExecutorService newBoundedExecutorService(
+      int maxConcurrency, boolean useVirtualThreads) {
+    return VirtualThreads.shouldUse(useVirtualThreads)
+        ? VirtualThreads.newBoundedVirtualThreadExecutor(maxConcurrency)
+        : Executors.newFixedThreadPool(maxConcurrency);
+  }
+
+  /**
+   * Creates the executor service used to run an unbounded number of tasks concurrently, either
+   * backed by virtual threads or by a cached pool of platform threads.
+   *
+   * @param useVirtualThreads whether virtual threads should be used, see {@link
+   *     ConfigurationService#useVirtualThreads()}
+   * @return the created {@link ExecutorService}
+   */
+  public static ExecutorService newUnboundedExecutorService(boolean useVirtualThreads) {
+    return VirtualThreads.shouldUse(useVirtualThreads)
+        ? VirtualThreads.newVirtualThreadPerTaskExecutor()
+        : Executors.newCachedThreadPool();
+  }
+
   /**
    * Uses cachingExecutorService from this manager. Use this only for tasks, that don't have dynamic
    * nature, in sense that won't grow with the number of inputs (thus kubernetes resources)
@@ -135,7 +166,8 @@ public ScheduledExecutorService scheduledExecutorService() {
   public synchronized void start(ConfigurationService configurationService) {
     if (!started) {
       this.configurationService = configurationService; // used to lazy init workflow executor
-      this.cachingExecutorService = Executors.newCachedThreadPool();
+      this.cachingExecutorService =
+          newUnboundedExecutorService(configurationService.useVirtualThreads());
       this.scheduledExecutorService = Executors.newScheduledThreadPool(0);
       this.executor = new InstrumentedExecutorService(configurationService.getExecutorService());
       started = true;

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

@@ -0,0 +1,178 @@
+/*
+ * Copyright Java Operator SDK Authors
+ *
+ * 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 io.javaoperatorsdk.operator.api.config;
+
+import java.lang.invoke.MethodHandle;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.MethodType;
+import java.util.List;
+import java.util.concurrent.AbstractExecutorService;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.javaoperatorsdk.operator.OperatorException;
+
+/**
+ * Creates the virtual thread based executors used when {@link
+ * ConfigurationService#useVirtualThreads()} is enabled.
+ *
+ * <p>The SDK is compiled for Java 17, in which virtual threads don't exist yet, so {@code
+ * Executors.newVirtualThreadPerTaskExecutor()} is looked up reflectively and is only available when
+ * the operator actually runs on Java 21 or later.
+ */
+final class VirtualThreads {
+
+  private static final Logger log = LoggerFactory.getLogger(VirtualThreads.class);
+
+  private static final MethodHandle NEW_VIRTUAL_THREAD_PER_TASK_EXECUTOR = lookupFactoryMethod();
+  private static final AtomicBoolean UNSUPPORTED_WARNING_LOGGED = new AtomicBoolean();
+
+  private VirtualThreads() {}
+
+  private static MethodHandle lookupFactoryMethod() {
+    try {
+      return MethodHandles.publicLookup()
+          .findStatic(
+              Executors.class,
+              "newVirtualThreadPerTaskExecutor",
+              MethodType.methodType(ExecutorService.class));
+    } catch (NoSuchMethodException | IllegalAccessException e) {
+      log.debug("Virtual threads are not available on this JVM", e);
+      return null;
+    }
+  }
+
+  /** Whether the JVM the operator runs on supports virtual threads, i.e. is Java 21 or later. */
+  static boolean isSupported() {
+    return NEW_VIRTUAL_THREAD_PER_TASK_EXECUTOR != null;
+  }
+
+  /**
+   * Whether virtual threads should effectively be used, i.e. they were requested through {@link
+   * ConfigurationService#useVirtualThreads()} <em>and</em> the JVM supports them. Requesting them
+   * on a JVM that doesn't support them is only warned about, so that the same configuration can be
+   * used regardless of the Java version the operator ends up running on, the only consequence being
+   * that platform threads are used instead. Concurrency limits are enforced either way.
+   */
+  static boolean shouldUse(boolean requested) {
+    if (!requested || isSupported()) {
+      return requested;
+    }
+    if (UNSUPPORTED_WARNING_LOGGED.compareAndSet(false, true)) {
+      log.warn(
+          "Virtual threads were requested but are not supported by the JVM in use (Java {}, Java 21"
+              + " or later is required). Falling back to platform threads.",
+          Runtime.version().feature());
+    }
+    return false;
+  }
+
+  /** An unbounded executor starting a new virtual thread for each submitted task. */
+  static ExecutorService newVirtualThreadPerTaskExecutor() {
+    if (!isSupported()) {
+      throw new OperatorException(
+          "Virtual threads are not supported by the JVM in use, Java 21 or later is required");
+    }
+    try {
+      return (ExecutorService) NEW_VIRTUAL_THREAD_PER_TASK_EXECUTOR.invokeExact();
+    } catch (Throwable e) {
+      throw new OperatorException("Couldn't create a virtual thread per task executor", e);
+    }
+  }
+
+  /**
+   * A virtual thread based executor executing at most {@code maxConcurrency} tasks at the same
+   * time, the equivalent of a fixed size platform thread pool.
+   */
+  static ExecutorService newBoundedVirtualThreadExecutor(int maxConcurrency) {
+    return new BoundedExecutorService(newVirtualThreadPerTaskExecutor(), maxConcurrency);
+  }
+
+  /**
+   * Limits how many of the tasks submitted to the wrapped executor run at the same time.
+   *
+   * <p>A thread is started for each task as soon as it is submitted, the task then waits for a
+   * permit before it actually runs. This only makes sense with virtual threads, which are cheap
+   * enough to be parked in large numbers, and has the property that submitting a task never blocks
+   * the submitting thread, just like queuing it on a fixed size platform thread pool wouldn't.
+   */
+  private static final class BoundedExecutorService extends AbstractExecutorService {
+
+    private final ExecutorService delegate;
+    private final Semaphore permits;
+
+    private BoundedExecutorService(ExecutorService delegate, int maxConcurrency) {
+      this.delegate = delegate;
+      // fair, so that tasks run roughly in submission order as they would on a thread pool
+      this.permits = new Semaphore(maxConcurrency, true);
+    }
+
+    @Override
+    public void execute(Runnable command) {
+      delegate.execute(
+          () -> {
+            try {
+              permits.acquire();
+            } catch (InterruptedException e) {
+              Thread.currentThread().interrupt();
+              // shutdownNow interrupted us before the task even started: cancel it so that whoever
+              // waits on the associated future isn't left hanging
+              if (command instanceof Future) {
+                ((Future<?>) command).cancel(false);
+              }
+              return;
+            }
+            try {
+              command.run();
+            } finally {
+              permits.release();
+            }
+          });
+    }
+
+    @Override
+    public void shutdown() {
+      delegate.shutdown();
+    }
+
+    @Override
+    public List<Runnable> shutdownNow() {
+      return delegate.shutdownNow();
+    }
+
+    @Override
+    public boolean isShutdown() {
+      return delegate.isShutdown();
+    }
+
+    @Override
+    public boolean isTerminated() {
+      return delegate.isTerminated();
+    }
+
+    @Override
+    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
+      return delegate.awaitTermination(timeout, unit);
+    }
+  }
+}

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverriderTest.java

@@ -44,7 +44,7 @@
   private static final Metrics METRICS = new Metrics() {};
 
   private static final LeaderElectionConfiguration LEADER_ELECTION_CONFIGURATION =
       new LeaderElectionConfiguration("foo", "fooNS");
 
   private static final Cloner CLONER =
       new Cloner() {
@@ -96,7 +96,7 @@
             .withConcurrentReconciliationThreads(25)
             .withMetrics(new Metrics() {})
             .withLeaderElectionConfiguration(
                 new LeaderElectionConfiguration("newLease", "newLeaseNS"))
             .withInformerStoppedHandler((informer, ex) -> {})
             .withReconciliationTerminationTimeout(Duration.ofSeconds(30))
             .build();
@@ -152,6 +152,13 @@
         .isEqualTo(14);
   }
 
+  @Test
+  void virtualThreadsAreDisabledByDefaultAndCanBeOverridden() {
+    assertThat(config.useVirtualThreads()).isFalse();
+    assertThat(new ConfigurationServiceOverrider(config).withUseVirtualThreads(true).build())
+        .returns(true, ConfigurationService::useVirtualThreads);
+  }
+
   @SuppressWarnings("rawtypes")
   @Test
   void dependentResourceFactoryDefaultsToTheSharedOneAndCanBeOverridden() {