feat: virtual thread support

docs/content/en/docs/documentation/eventing.md

@@ -222,6 +222,35 @@ is similar to `PerResourcePollingEventSource` except that, contrary to that even
 doesn't poll a specific API separately per resource, but periodically and independently of
 actually observed primary resources.
 
+#### Threading of the polling event sources
+
+Both polling event sources schedule their polls on an executor the operator shares between all of
+its polling event sources. A poll therefore only starts once one of that executor's threads is
+free, so if your operator registers many polling event sources, or if fetching your external
+resources is slow, size the pool accordingly with
+`ConfigurationServiceOverrider.withConcurrentScheduledTaskThreads` (4 threads by default):
+
+```java
+Operator operator = new Operator(overrider -> overrider.withConcurrentScheduledTaskThreads(20));
+```
+
+With virtual threads enabled, see [virtual threads](../operations/configuration/#virtual-threads), the
+polls run on virtual threads, the executor only keeping one platform thread to do the timing; the
+number of threads then caps how many polls run at the same time.
+
+Retried and rescheduled reconciliations are triggered on a separate executor, so a slow poll can
+never delay them. It is sized with `withConcurrentRetryAndRescheduleThreads` (2 threads by
+default); few threads are needed there since triggering a reconciliation only enqueues an event
+for one of the reconciliation threads to pick up.
+
+Use `ConfigurationServiceOverrider.withScheduledExecutorService` to replace the polling executor
+altogether. The operator still manages the lifecycle of that executor: it shuts it down when the
+operator is stopped, so don't shut it down yourself.
+
+Use the `withExecutorService` method of the event source's own configuration builder to poll a
+single event source on an executor of its own. An executor provided that way is not managed by the
+operator: it is your responsibility to shut it down.
+
 #### Inbound event sources
 
 [SimpleInboundEventSource](https://github.com/operator-framework/java-operator-sdk/blob/main/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/inbound/SimpleInboundEventSource.java)

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

@@ -23,6 +23,55 @@ 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.
+
+Things to keep in mind:
+
+- Virtual threads are only supported on **Java 25 or later**. Before Java 25 a virtual thread pins
+  its carrier thread while it is inside a `synchronized` block, which can starve the carrier pool
+  and even deadlock the operator on startup. [JEP 491](https://openjdk.org/jeps/491), delivered in
+  Java 25, removed that pinning. This is checked at runtime: on an older JVM, including the ones
+  where virtual threads exist (Java 21 to 24), a warning is logged and platform threads are used
+  instead, so the same configuration works on any Java version the operator runs on.
+- The scheduled tasks (the polling event sources, and the retried and rescheduled reconciliations)
+  run on virtual threads too. The JDK has no scheduled executor backed by virtual threads, so the
+  default ones keep a single platform thread for the timing only and run each task they fire on a
+  virtual thread. `withConcurrentScheduledTaskThreads(int)` and
+  `withConcurrentRetryAndRescheduleThreads(int)` still cap how many of those tasks run at the same
+  time.
+- With virtual threads, which are always daemon threads, the framework keeps one idle platform
+  thread (`josdk-keep-alive`) running until the operator is stopped, so that an operator whose
+  `main` method returns after starting it doesn't let the JVM exit.
+- A custom `ExecutorService` provided through `withExecutorService(...)` or
+  `withWorkflowExecutorService(...)` is always used as is, the flag has no effect on it.
+- The Kubernetes client the framework creates when none is provided also switches its internal
+  task executor (used to dispatch informer events to their handlers and to deliver watch events)
+  to virtual threads. A client you provide through `withKubernetesClient(...)` is used as is: to
+  get the same behavior, configure it yourself, e.g. with
+  `new KubernetesClientBuilder().withTaskExecutor(Executors.newVirtualThreadPerTaskExecutor())`
+  (the client doesn't shut down an executor passed that way, use `withTaskExecutorSupplier(...)`
+  if it should be shut down when the client is closed).
+  Either way, the blocking calls your reconciler makes through the client work well on virtual
+  threads without any client-side change.
+
 ## 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:
@@ -267,6 +316,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 (only supported on Java 25+, platform threads are used on older versions) |
 | `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 |
 
@@ -283,6 +333,13 @@ All operator-level keys are prefixed with `josdk.`.
 |---|---|---|
 | `josdk.workflow.executor-threads` | `Integer` | Thread pool size for workflow execution |
 
+#### Scheduled Tasks
+
+| Key | Type | Description |
+|---|---|---|
+| `josdk.scheduled-tasks.concurrent-threads` | `Integer` | Thread pool size for the operator's scheduled tasks, i.e. the polling event sources |
+| `josdk.retry-and-reschedule.concurrent-threads` | `Integer` | Thread pool size for triggering retried and rescheduled reconciliations |
+
 #### Informer
 
 | Key | Type | Description |

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/ExecutorUtilsInternal.java

@@ -0,0 +1,69 @@
+/*
+ * 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;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class ExecutorUtilsInternal {
+
+  // prevent instantiation of util class
+  private ExecutorUtilsInternal() {}
+
+  /**
+   * Creates a {@link ThreadFactory} producing daemon threads named after the specified prefix.
+   * Daemon threads don't keep the JVM alive if the {@link Operator} is never stopped, and naming
+   * them makes the pool they belong to identifiable in thread dumps.
+   *
+   * @param namePrefix the prefix the created threads are named after
+   * @return a {@link ThreadFactory} creating named daemon threads
+   */
+  public static ThreadFactory daemonThreadFactory(String namePrefix) {
+    final var defaultThreadFactory = Executors.defaultThreadFactory();
+    final var counter = new AtomicLong();
+    return runnable -> {
+      final var thread = defaultThreadFactory.newThread(runnable);
+      thread.setName(namePrefix + "-" + counter.incrementAndGet());
+      thread.setDaemon(true);
+      return thread;
+    };
+  }
+
+  /**
+   * Creates the kind of {@link ScheduledExecutorService} the operator runs its scheduled (i.e.
+   * periodic or delayed) tasks on: one whose threads are daemon threads named after the specified
+   * prefix, and which doesn't let the tasks that were scheduled for later delay its shutdown.
+   *
+   * @param corePoolSize the number of threads to keep in the pool
+   * @param threadNamePrefix the prefix the threads of the pool are named after
+   * @return a {@link ScheduledExecutorService} to run scheduled tasks on
+   */
+  public static ScheduledExecutorService daemonScheduledThreadPool(
+      int corePoolSize, String threadNamePrefix) {
+    final var executor =
+        new ScheduledThreadPoolExecutor(corePoolSize, daemonThreadFactory(threadNamePrefix));
+    // tasks that are scheduled far out (a reconciliation rescheduled in an hour, say) would
+    // otherwise keep the pool from terminating until the graceful shutdown timeout expires
+    executor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
+    // cancelled tasks are frequent (every retry that is superseded by a new event cancels one) and
+    // would otherwise be retained until their delay elapses
+    executor.setRemoveOnCancelPolicy(true);
+    return executor;
+  }
+}