feat: virtual thread support

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

@@ -222,6 +222,37 @@ 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
+
+Each polling event source polls on a scheduled executor of its own, so a slow poll of one event
+source never delays the polls of the others. Unless you provide one, the event source creates a
+single daemon thread executor when it is started and shuts it down when it is stopped. With a single
+thread the polls of a source run one after the other. A `PollingEventSource` runs a single periodic
+task, so more threads would not help it. A `PerResourcePollingEventSource` however polls each
+registered resource on its own, so if it has many registered resources or fetches slowly, it can be
+given an executor with more threads using the `withExecutorService` method of its configuration
+builder:
+
+```java
+var config =
+    new PerResourcePollingConfigurationBuilder<>(this::fetchResources, Duration.ofSeconds(5))
+        .withExecutorService(Executors.newScheduledThreadPool(4))
+        .build();
+```
+
+An executor provided that way is not managed by the operator: it is your responsibility to shut it
+down. It is only ever used as is, the virtual threads flag has no effect on it. You can also share
+it between multiple event sources deliberately.
+
+With virtual threads enabled, see [virtual threads](../operations/configuration/#virtual-threads),
+the executor the event source creates runs the polls on virtual threads, only keeping one platform
+thread to do the timing. The number of polls running at the same time is then not limited.
+
+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.
+
 #### 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,60 @@ 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 polls of the polling event sources, which each create an executor of
+  their own unless one is provided, 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. The
+  number of those tasks running at the same time is not limited, so
+  `withConcurrentRetryAndRescheduleThreads(int)` only applies to platform threads.
+- Virtual threads are always daemon threads, so unlike the threads of the platform thread pools they
+  don't keep the JVM running: if the `main` method of your application returns after starting the
+  operator and nothing else keeps the JVM alive, call `Operator.keepAlive()`, see
+  [Keeping the JVM Alive](#keeping-the-jvm-alive).
+- A custom `ExecutorService` provided through `withExecutorService(...)` or
+  `withWorkflowExecutorService(...)` is always used as is, the flag has no effect on it. The same
+  goes for `withRetryAndRescheduleExecutorService(...)`. Note however that overriding the flag, or
+  `withConcurrentRetryAndRescheduleThreads(int)`, of a configuration that provides its own retry and
+  reschedule executor replaces that executor with the default one, so provide it again with
+  `withRetryAndRescheduleExecutorService(...)` in that case.
+- 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:
@@ -285,6 +339,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 |
 
@@ -301,6 +356,12 @@ All operator-level keys are prefixed with `josdk.`.
 |---|---|---|
 | `josdk.workflow.executor-threads` | `Integer` | Thread pool size for workflow execution |
 
+#### Retry and Reschedule
+
+| Key | Type | Description |
+|---|---|---|
+| `josdk.retry-and-reschedule.concurrent-threads` | `Integer` | Thread pool size for triggering retried and rescheduled reconciliations |
+
 #### Informer
 
 | Key | Type | Description |

docs/content/en/docs/migration/v5-7-migration.md

@@ -78,3 +78,35 @@ on the classpath next to your implementation. This is especially important if yo
 based client (`kubernetes-httpclient-vertx`), since Vert.x 4 and Vert.x 5 cannot be used together.
 If you want to keep the Vert.x 4 based client, use it in place of `kubernetes-httpclient-jdk` in the
 snippet above.
+
+## Executors of the polling and timer event sources
+
+The polling event sources and the internal `TimerEventSource` no longer use a `java.util.Timer`, and
+`PerResourcePollingEventSource` no longer creates its executor when its configuration is built. They
+run on scheduled executors, see [virtual threads](../documentation/operations/configuration/#virtual-threads)
+and [Threading of the polling event sources](../documentation/eventing/#threading-of-the-polling-event-sources).
+This is transparent unless your operator provides an executor to a polling event source, or relies
+on one of the following:
+
+- A `ScheduledExecutorService` provided with `withExecutorService(...)` to a
+  `PerResourcePollingConfigurationBuilder` was shut down with `shutdownNow()` when the event source
+  was stopped. It is not anymore: **an executor you provide is yours to shut down**, like it is for
+  `PollingConfigurationBuilder`. If you relied on the event source for this, shut it down yourself
+  after stopping the operator, otherwise its threads can keep the JVM from exiting if they are not
+  daemon threads.
+- `PerResourcePollingConfiguration.executorService()` returns `null` when no executor was provided,
+  it used to be a `ScheduledThreadPoolExecutor` with a single non-daemon thread created by the
+  configuration. The event source now creates a single daemon thread executor, or one running on
+  virtual threads if `useVirtualThreads` is enabled, when it starts, shuts it down when it stops, and
+  creates a new one if it is started again. `PerResourcePollingConfiguration.DEFAULT_EXECUTOR_THREAD_NUMBER`
+  is deprecated, it isn't used anymore.
+- Each polling event source has an executor of its own: they don't share threads, and the number of
+  threads of a source is not configurable other than by providing an executor.
+- `new PollingEventSource<>(Class, PollingConfiguration)` is deprecated. Use the constructor that
+  takes the `EventSourceContext` as well, `new PollingEventSource<>(Class, EventSourceContext,
+  PollingConfiguration)`, otherwise the event source cannot know about the `useVirtualThreads`
+  setting and polls on a platform thread.
+- `TimerEventSource(String, boolean)` is deprecated, use the constructor that also takes an
+  executor supplier, `null` to have the event source create an executor of its own.
+- `ExecutorServiceManager.scheduledExecutorService()` is deprecated. The framework doesn't use it
+  anymore, it creates the executor on first use, and shuts it down when the manager is stopped.

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

@@ -0,0 +1,77 @@
+/*
+ * 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.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class ExecutorUtilsInternal {
+
+  private static final long SCHEDULED_THREAD_KEEP_ALIVE_SECONDS = 60;
+
+  // 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 maximal number of threads of the pool, which are created on demand and
+   *     let go if idle for a while
+   * @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);
+    // the threads are created as tasks get scheduled, not as the load requires, so without this
+    // they would all stay around idle for good; the pool always keeps a thread for delayed tasks
+    executor.setKeepAliveTime(SCHEDULED_THREAD_KEEP_ALIVE_SECONDS, TimeUnit.SECONDS);
+    executor.allowCoreThreadTimeOut(true);
+    return executor;
+  }
+}