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;
@@ -203,6 +202,11 @@ default boolean checkCRDAndValidateLocalModel() {
    * The number of threads the operator can spin out to dispatch reconciliation requests to
    * reconcilers with the default executors
    *
+   * <p>This is a concurrency limit and applies regardless of whether the default executor is backed
+   * by platform or by virtual threads, see {@link #useVirtualThreads()}: with virtual threads it
+   * caps how many reconciliations run at the same time rather than the size of a thread pool. Since
+   * virtual threads are cheap, the limit can be set considerably higher when they are enabled.
+   *
    * @return the number of concurrent reconciliation threads
    */
   default int concurrentReconciliationThreads() {
@@ -213,6 +217,12 @@ default int concurrentReconciliationThreads() {
    * Number of threads the operator can spin out to be used in the workflows with the default
    * executor.
    *
+   * <p>This is a concurrency limit and applies regardless of whether the default executor is backed
+   * by platform or by virtual threads, see {@link #useVirtualThreads()}: with virtual threads it
+   * caps how many dependent resources are processed at the same time rather than the size of a
+   * thread pool. Since virtual threads are cheap, the limit can be set considerably higher when
+   * they are enabled.
+   *
    * @return the maximum number of concurrent workflow threads
    */
   default int concurrentWorkflowExecutorThreads() {
@@ -228,6 +238,34 @@ default Metrics getMetrics() {
     return Metrics.NOOP;
   }
 
+  /**
+   * Whether the framework should run the tasks it executes concurrently &mdash; reconciliations,
+   * dependent workflows and internal housekeeping such as starting the informers &mdash; 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 +274,8 @@ default Metrics getMetrics() {
    *     processing
    */
   default ExecutorService getExecutorService() {
-    return Executors.newFixedThreadPool(concurrentReconciliationThreads());
+    return ExecutorServiceManager.newBoundedExecutorService(
+        concurrentReconciliationThreads(), useVirtualThreads());
   }
 
   /**
@@ -246,7 +285,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,10 @@ 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());
+      // stays on platform threads even when virtual threads are requested: there is no virtual
+      // thread backed ScheduledExecutorService in the JDK, see VirtualThreads
       this.scheduledExecutorService = Executors.newScheduledThreadPool(0);
       this.executor = new InstrumentedExecutorService(configurationService.getExecutorService());
       started = true;