feat: virtual thread support

bootstrapper-maven-plugin/pom.xml

@@ -22,7 +22,7 @@
   <parent>
     <groupId>io.javaoperatorsdk</groupId>
     <artifactId>java-operator-sdk</artifactId>
-    <version>5.6.1-SNAPSHOT</version>
+    <version>999-SNAPSHOT</version>
   </parent>
 
   <artifactId>bootstrapper</artifactId>

caffeine-bounded-cache-support/pom.xml

@@ -21,7 +21,7 @@
   <parent>
     <groupId>io.javaoperatorsdk</groupId>
     <artifactId>java-operator-sdk</artifactId>
-    <version>5.6.1-SNAPSHOT</version>
+    <version>999-SNAPSHOT</version>
   </parent>
 
   <artifactId>caffeine-bounded-cache-support</artifactId>

docs/content/en/docs/documentation/dependent-resource-and-workflows/dependent-resources.md

@@ -531,6 +531,34 @@ samples [here](https://github.com/java-operator-sdk/java-operator-sdk/tree/main/
    in [related integration test](https://github.com/operator-framework/java-operator-sdk/blob/main/operator-framework/src/test/java/io/javaoperatorsdk/operator/workflow/orderedmanageddependent/ConfigMapDependentResource2.java)
    .
 
+### Adding Common Metadata to All Managed Resources (Desired State Aspects)
+
+Operators often need to mark every resource they manage in a uniform way, for example with a
+`app.kubernetes.io/managed-by` label, so that these resources can easily be identified, selected or
+garbage-collected later on. Instead of repeating that logic in every `desired()` implementation, a
+`DesiredStateAspect` can be registered once, at the operator level, and is then applied to the
+desired state of every Kubernetes dependent resource managed by the operator:
+
+```java
+Operator operator = new Operator(overrider -> overrider
+    .withDesiredStateAspects(List.of(
+        (desired, dependentResource, context) -> desired.getMetadata().getLabels()
+            .put("app.kubernetes.io/managed-by", "my-operator"))));
+```
+
+Aspects are applied, in registration order, right after the desired state has been computed and
+before the desired state is matched against the actual resource, created or updated. As a
+consequence, the metadata added by an aspect is part of the desired state proper: if it is removed
+from the actual resource, or if the aspect itself changes, the associated secondary resources are
+updated accordingly on the next reconciliation.
+
+Since the desired state is computed at most once per reconciliation and cached in the `Context`,
+aspects are called at most once per dependent resource and reconciliation. They are only called for
+dependent resources whose desired state is a `HasMetadata`, meaning that external (non-Kubernetes)
+dependent resources are left untouched. Implementations are expected to modify the provided desired
+state in place and need to be thread-safe as they can be called concurrently for different primary
+resources.
+
 ## "Read-only" Dependent Resources vs. Event Source
 
 See Integration test for a read-only

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

@@ -348,6 +348,30 @@ See
 also [CaffeineBoundedItemStores](https://github.com/operator-framework/java-operator-sdk/blob/main/caffeine-bounded-cache-support/src/main/java/io/javaoperatorsdk/operator/processing/event/source/cache/CaffeineBoundedItemStores.java)
 for more details.
 
+### Removing the Namespace Index
+
+Informers keep an index from namespace to the resources cached for it. JOSDK never reads that index,
+only the ones registered explicitly through `IndexerResourceCache.addIndexers(..)`, so it can be
+removed to save an entry per cached resource:
+
+```java
+@ControllerConfiguration(informer = @Informer(withoutNamespaceIndex = true))
+public class MyReconciler implements Reconciler<MyCustomResource> { }
+```
+
+The same option is available on `InformerEventSourceConfiguration.Builder` for event sources, and on
+`InformerConfiguration.Builder`.
+
+This matters most for informers that cache a large number of resources, and in particular together
+with a custom [item store](#bounded-caches-for-informers) that keeps only a reduced form of each
+resource: the index is keyed independently of what the store does with the resource itself, so
+shrinking what is cached does not shrink the index.
+
+Note that this takes part in the informer pool identity described below: an event source that
+removes the index does not share an informer with one that keeps it. If two event sources watch the
+same resource type and only one of them sets the option, they end up with two informers, and two
+caches of that resource type, which can cost far more memory than the index ever did.
+
 ### Sharing Informers Between Controllers (Informer Pool)
 
 {{% alert title="Experimental" color="warning" %}}
@@ -374,7 +398,10 @@ Two event sources share an informer when their effective informer configuration
 - the resource type (or the group/version/kind for generic resources),
 - the watched namespace,
 - the label, field and shard selectors,
-- the configured [item store](#bounded-caches-for-informers).
+- the configured [item store](#bounded-caches-for-informers),
+- whether the [namespace index is removed](#removing-the-namespace-index): it cannot be present for
+  one event source and absent for another on one shared informer, so event sources that disagree on
+  it are backed by separate informers.
 
 The `informerListLimit` is intentionally *not* part of this identity: if two otherwise-equivalent
 event sources request a different list limit, the existing informer is reused (a warning is logged

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

@@ -23,6 +23,47 @@ 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 are officially supported on **Java 25 or later**. They exist from Java 21 on and
+  the flag does enable them there, but before Java 25 a virtual thread pins its carrier thread
+  while it is inside a `synchronized` block, which can starve the carrier pool. [JEP
+  491](https://openjdk.org/jeps/491), delivered in Java 25, removed that pinning, so this is the
+  baseline the framework supports. On a JVM without virtual threads at all (below Java 21) 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.
+- 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:
@@ -265,6 +306,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 (officially supported on Java 25+ 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 |
 
@@ -356,6 +398,7 @@ All controller-level keys are prefixed with `josdk.controller.<controller-name>.
 | `josdk.controller.<name>.informer.label-selector` | `String` | Label selector for the primary resource informer (alias for `label-selector`) |
 | `josdk.controller.<name>.informer.shard-selector` | `String` | Shard selector for the primary resource informer (alias for `shard-selector`) |
 | `josdk.controller.<name>.informer.list-limit` | `Long` | Page size for paginated informer list requests; omit for no pagination |
+| `josdk.controller.<name>.informer.without-namespace-index` | `Boolean` | Removes the namespace index the informer maintains; defaults to `false`. See [Removing the Namespace Index](../eventing#removing-the-namespace-index) |
 
 #### Retry
 

micrometer-support/pom.xml

@@ -21,7 +21,7 @@
   <parent>
     <groupId>io.javaoperatorsdk</groupId>
     <artifactId>java-operator-sdk</artifactId>
-    <version>5.6.1-SNAPSHOT</version>
+    <version>999-SNAPSHOT</version>
   </parent>
 
   <artifactId>micrometer-support</artifactId>

migration/pom.xml

@@ -21,7 +21,7 @@
   <parent>
     <groupId>io.javaoperatorsdk</groupId>
     <artifactId>java-operator-sdk</artifactId>
-    <version>5.6.1-SNAPSHOT</version>
+    <version>999-SNAPSHOT</version>
   </parent>
 
   <artifactId>migration</artifactId>

operator-framework-bom/pom.xml

@@ -21,7 +21,7 @@
 
   <groupId>io.javaoperatorsdk</groupId>
   <artifactId>operator-framework-bom</artifactId>
-  <version>5.6.1-SNAPSHOT</version>
+  <version>999-SNAPSHOT</version>
   <packaging>pom</packaging>
   <name>Operator SDK - Bill of Materials</name>
   <description>Java SDK for implementing Kubernetes operators</description>

operator-framework-core/pom.xml

@@ -21,7 +21,7 @@
   <parent>
     <groupId>io.javaoperatorsdk</groupId>
     <artifactId>java-operator-sdk</artifactId>
-    <version>5.6.1-SNAPSHOT</version>
+    <version>999-SNAPSHOT</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 
@@ -128,7 +128,7 @@
         <version>${git-commit-id-maven-plugin.version}</version>
         <configuration>
           <generateGitPropertiesFile>true</generateGitPropertiesFile>
-          <generateGitPropertiesFilename>${project.build.outputDirectory}/version.properties</generateGitPropertiesFilename>
+          <generateGitPropertiesFilename>${project.build.outputDirectory}/operator-sdk-version.properties</generateGitPropertiesFilename>
           <includeOnlyProperties>
             <includeOnlyProperty>^git.build.time$</includeOnlyProperty>
             <includeOnlyProperty>^git.commit.id.(abbrev|full)$</includeOnlyProperty>

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

@@ -16,10 +16,10 @@
 package io.javaoperatorsdk.operator.api.config;
 
 import java.time.Duration;
+import java.util.List;
 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;
@@ -41,6 +41,7 @@
 import io.javaoperatorsdk.operator.api.reconciler.Experimental;
 import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResourceFactory;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.DesiredStateAspect;
 import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependent;
 import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResource;
 import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResourceConfig;
@@ -153,6 +154,12 @@ public <R extends HasMetadata> R clone(R object) {
    * io.javaoperatorsdk.operator.Operator#Operator(Consumer)}, passing your custom instance with
    * {@link ConfigurationServiceOverrider#withKubernetesClient(KubernetesClient)}.
    *
+   * <p>When {@link #useVirtualThreads()} is enabled (and supported by the JVM), the default client
+   * runs its internal asynchronous tasks, such as dispatching informer events to their handlers and
+   * delivering watch events, on virtual threads instead of its default cached platform thread pool.
+   * A client provided by overriding this method or through {@link
+   * ConfigurationServiceOverrider#withKubernetesClient(KubernetesClient)} is used as is.
+   *
    * <p><em>NOTE:</em> It is strongly suggested that implementors override this method since the
    * default implementation creates a new {@link KubernetesClient} instance each time this method is
    * called.
@@ -161,13 +168,17 @@ public <R extends HasMetadata> R clone(R object) {
    * @since 4.4.0
    */
   default KubernetesClient getKubernetesClient() {
-    return new KubernetesClientBuilder()
-        .withConfig(
-            new ConfigBuilder(Config.autoConfigure(null))
-                .withMaxConcurrentRequests(DEFAULT_MAX_CONCURRENT_REQUEST)
-                .build())
-        .withKubernetesSerialization(new KubernetesSerialization())
-        .build();
+    final var builder =
+        new KubernetesClientBuilder()
+            .withConfig(
+                new ConfigBuilder(Config.autoConfigure(null))
+                    .withMaxConcurrentRequests(DEFAULT_MAX_CONCURRENT_REQUEST)
+                    .build())
+            .withKubernetesSerialization(new KubernetesSerialization());
+    if (VirtualThreads.shouldUse(useVirtualThreads())) {
+      builder.withTaskExecutorSupplier(VirtualThreads.newKubernetesClientTaskExecutorSupplier());
+    }
+    return builder.build();
   }
 
   /**
@@ -201,6 +212,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() {
@@ -211,6 +227,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() {
@@ -226,6 +248,42 @@ 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>Officially supported on Java 25 or later. Virtual threads exist as of Java 21 and are used
+   * there as well, but before Java 25 a virtual thread pins its carrier thread while inside a
+   * {@code synchronized} block, which <a href="https://openjdk.org/jeps/491">JEP 491</a> removed in
+   * Java 25. On a JVM without virtual threads at all, 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. The same goes for the {@link KubernetesClient}:
+   * the default client created by {@link #getKubernetesClient()} then also runs its internal
+   * asynchronous tasks (informer event dispatching, watch event delivery) on virtual threads, while
+   * a custom client, e.g. provided through {@link
+   * ConfigurationServiceOverrider#withKubernetesClient(KubernetesClient)}, keeps whatever task
+   * executor it was built with. The blocking calls a reconciler performs through the client don't
+   * need any of this: they already park the calling virtual thread rather than its carrier.
+   *
+   * @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
@@ -234,7 +292,8 @@ default Metrics getMetrics() {
    *     processing
    */
   default ExecutorService getExecutorService() {
-    return Executors.newFixedThreadPool(concurrentReconciliationThreads());
+    return ExecutorServiceManager.newBoundedExecutorService(
+        concurrentReconciliationThreads(), useVirtualThreads());
   }
 
   /**
@@ -244,7 +303,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());
   }
 
   /**
@@ -539,4 +599,18 @@ default InformerPool informerPool() {
     pool.setConfigurationService(this);
     return pool;
   }
+
+  /**
+   * Retrieves the {@link DesiredStateAspect}s applied to the desired state of all the Kubernetes
+   * dependent resources managed by the operator. Aspects are applied in the order in which they are
+   * returned, right after the desired state has been computed, and are typically used to add common
+   * metadata (such as a label identifying the operator managing the resource) to all the resources
+   * the operator creates or updates.
+   *
+   * @return the list of aspects to apply to computed desired states, empty by default
+   * @since 5.6.0
+   */
+  default List<DesiredStateAspect> desiredStateAspects() {
+    return List.of();
+  }
 }