From d98eb7b474b35996f827a15cf19847fdc6fadfef Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Attila=20M=C3=A9sz=C3=A1ros?= Date: Mon, 7 Sep 2026 15:57:55 +0200 Subject: [PATCH 1/6] feat: virtual thread support MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Attila Mészáros --- .../documentation/operations/configuration.md | 29 +++ .../api/config/ConfigurationService.java | 35 +++- .../config/ConfigurationServiceOverrider.java | 19 ++ .../api/config/ExecutorServiceManager.java | 34 +++- .../operator/api/config/VirtualThreads.java | 178 ++++++++++++++++++ .../ConfigurationServiceOverriderTest.java | 7 + .../api/config/VirtualThreadsTest.java | 162 ++++++++++++++++ .../operator/config/loader/ConfigLoader.java | 4 + .../VirtualThreadsCustomResource.java | 30 +++ .../virtualthreads/VirtualThreadsIT.java | 92 +++++++++ .../VirtualThreadsTestReconciler.java | 84 +++++++++ .../config/loader/ConfigLoaderTest.java | 2 + 12 files changed, 672 insertions(+), 4 deletions(-) create mode 100644 operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java create mode 100644 operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VirtualThreadsTest.java create mode 100644 operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsCustomResource.java create mode 100644 operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsIT.java create mode 100644 operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsTestReconciler.java diff --git a/docs/content/en/docs/documentation/operations/configuration.md b/docs/content/en/docs/documentation/operations/configuration.md index 9e2576cbfc..b5a71bb126 100644 --- a/docs/content/en/docs/documentation/operations/configuration.md +++ b/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 | diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java index 35f46e5019..e3454e6c98 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java +++ b/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. + * + *

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 + * not 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. + * + *

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. + * + *

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()); } /** diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java index 2cf6540af0..b3ae079561 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java +++ b/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) { diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManager.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManager.java index cdcafcaa46..2176cb0fab 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManager.java +++ b/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; diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java new file mode 100644 index 0000000000..dd7a8a80a0 --- /dev/null +++ b/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. + * + *

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()} and 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. + * + *

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 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); + } + } +} diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverriderTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverriderTest.java index aec8381135..0247e677c6 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverriderTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverriderTest.java @@ -152,6 +152,13 @@ void threadCountConfiguredProperly() { .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() { diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VirtualThreadsTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VirtualThreadsTest.java new file mode 100644 index 0000000000..60d7e26f54 --- /dev/null +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VirtualThreadsTest.java @@ -0,0 +1,162 @@ +/* + * 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.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.IntStream; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.EnabledForJreRange; +import org.junit.jupiter.api.condition.JRE; + +import static org.assertj.core.api.Assertions.assertThat; + +class VirtualThreadsTest { + + private static final int MAX_CONCURRENCY = 3; + private static final int TASK_NUMBER = 20; + private static final int TIMEOUT_SECONDS = 30; + + @Test + void usesPlatformThreadPoolWhenVirtualThreadsAreNotRequested() throws Exception { + var executor = ExecutorServiceManager.newBoundedExecutorService(MAX_CONCURRENCY, false); + try { + assertThat(executor).isInstanceOf(ThreadPoolExecutor.class); + assertThat(executor.submit(VirtualThreadsTest::onVirtualThread).get()).isFalse(); + } finally { + executor.shutdownNow(); + } + } + + @Test + @EnabledForJreRange(min = JRE.JAVA_21) + void usesVirtualThreadsWhenRequested() throws Exception { + var virtualExecutor = ExecutorServiceManager.newBoundedExecutorService(MAX_CONCURRENCY, true); + try { + assertThat(virtualExecutor.submit(VirtualThreadsTest::onVirtualThread).get()).isTrue(); + } finally { + virtualExecutor.shutdownNow(); + } + + var unbounded = ExecutorServiceManager.newUnboundedExecutorService(true); + try { + assertThat(unbounded.submit(VirtualThreadsTest::onVirtualThread).get()).isTrue(); + } finally { + unbounded.shutdownNow(); + } + } + + @Test + @EnabledForJreRange(min = JRE.JAVA_21) + void boundedVirtualThreadExecutorRespectsTheConfiguredConcurrency() throws Exception { + var executor = ExecutorServiceManager.newBoundedExecutorService(MAX_CONCURRENCY, true); + try { + final var running = new AtomicInteger(); + final var maxObservedConcurrency = new AtomicInteger(); + final var done = new CountDownLatch(TASK_NUMBER); + // each task gets its own (virtual) thread, only the number of concurrently running ones is + // capped + final Set usedThreads = ConcurrentHashMap.newKeySet(); + + IntStream.range(0, TASK_NUMBER) + .forEach( + i -> + executor.execute( + () -> { + usedThreads.add(Thread.currentThread()); + maxObservedConcurrency.accumulateAndGet( + running.incrementAndGet(), Math::max); + try { + Thread.sleep(50); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + running.decrementAndGet(); + done.countDown(); + } + })); + + assertThat(done.await(TIMEOUT_SECONDS, TimeUnit.SECONDS)).isTrue(); + assertThat(maxObservedConcurrency).hasValue(MAX_CONCURRENCY); + assertThat(usedThreads).hasSize(TASK_NUMBER); + } finally { + executor.shutdownNow(); + } + } + + @Test + @EnabledForJreRange(min = JRE.JAVA_21) + void invokeAllIsBoundedTooSinceItIsUsedToStartTheEventSources() { + var executor = ExecutorServiceManager.newBoundedExecutorService(MAX_CONCURRENCY, true); + try { + final var running = new AtomicInteger(); + final var maxObservedConcurrency = new AtomicInteger(); + + ExecutorServiceManager.executeAndWaitForAllToComplete( + IntStream.range(0, TASK_NUMBER).boxed(), + i -> { + maxObservedConcurrency.accumulateAndGet(running.incrementAndGet(), Math::max); + try { + Thread.sleep(50); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + running.decrementAndGet(); + } + return null; + }, + i -> "task-" + i, + executor); + + assertThat(maxObservedConcurrency).hasValue(MAX_CONCURRENCY); + } finally { + executor.shutdownNow(); + } + } + + @Test + @EnabledForJreRange(min = JRE.JAVA_21) + void shutdownTerminatesOnceTheAlreadySubmittedTasksAreDone() throws Exception { + ExecutorService executor = + ExecutorServiceManager.newBoundedExecutorService(MAX_CONCURRENCY, true); + final var done = new CountDownLatch(TASK_NUMBER); + + IntStream.range(0, TASK_NUMBER).forEach(i -> executor.execute(done::countDown)); + executor.shutdown(); + + assertThat(executor.awaitTermination(TIMEOUT_SECONDS, TimeUnit.SECONDS)).isTrue(); + assertThat(executor.isShutdown()).isTrue(); + assertThat(done.getCount()).isZero(); + } + + /** + * {@code Thread.isVirtual} only exists as of Java 21 while the tests are compiled for Java 17, + * hence the reflective call. + */ + static boolean onVirtualThread() { + try { + return (Boolean) Thread.class.getMethod("isVirtual").invoke(Thread.currentThread()); + } catch (ReflectiveOperationException e) { + return false; + } + } +} diff --git a/operator-framework/src/main/java/io/javaoperatorsdk/operator/config/loader/ConfigLoader.java b/operator-framework/src/main/java/io/javaoperatorsdk/operator/config/loader/ConfigLoader.java index bc9a4ecbbb..f15ffaf78b 100644 --- a/operator-framework/src/main/java/io/javaoperatorsdk/operator/config/loader/ConfigLoader.java +++ b/operator-framework/src/main/java/io/javaoperatorsdk/operator/config/loader/ConfigLoader.java @@ -82,6 +82,10 @@ public static ConfigLoader getDefault() { "close-client-on-stop", Boolean.class, ConfigurationServiceOverrider::withCloseClientOnStop), + new ConfigBinding<>( + "use-virtual-threads", + Boolean.class, + ConfigurationServiceOverrider::withUseVirtualThreads), new ConfigBinding<>( "informer.stop-on-error-during-startup", Boolean.class, diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsCustomResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsCustomResource.java new file mode 100644 index 0000000000..be086a5f1b --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsCustomResource.java @@ -0,0 +1,30 @@ +/* + * 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.baseapi.virtualthreads; + +import io.fabric8.kubernetes.api.model.Namespaced; +import io.fabric8.kubernetes.client.CustomResource; +import io.fabric8.kubernetes.model.annotation.Group; +import io.fabric8.kubernetes.model.annotation.Kind; +import io.fabric8.kubernetes.model.annotation.ShortNames; +import io.fabric8.kubernetes.model.annotation.Version; + +@Group("sample.javaoperatorsdk") +@Version("v1") +@Kind("VirtualThreadsCustomResource") +@ShortNames("vtc") +public class VirtualThreadsCustomResource extends CustomResource + implements Namespaced {} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsIT.java new file mode 100644 index 0000000000..fbb38042aa --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsIT.java @@ -0,0 +1,92 @@ +/* + * 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.baseapi.virtualthreads; + +import java.util.concurrent.TimeUnit; +import java.util.stream.IntStream; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import io.fabric8.kubernetes.api.model.ObjectMeta; +import io.javaoperatorsdk.annotation.Sample; +import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension; + +import static io.javaoperatorsdk.operator.baseapi.virtualthreads.VirtualThreadsTestReconciler.onVirtualThread; +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +@Sample( + tldr = "Running reconciliations on virtual threads", + description = + """ + Demonstrates how to make the framework execute its concurrent work on virtual threads by \ + simply setting a flag on the ConfigurationService. Virtual threads make the blocking calls \ + a reconciler typically performs much cheaper, while the configured concurrency limits are \ + still enforced: the test verifies that the reconciler runs on virtual threads and that no \ + more than the configured number of reconciliations happen at the same time. + """) +class VirtualThreadsIT { + + static final int CONCURRENT_RECONCILIATION_THREADS = 2; + static final int NUMBER_OF_RESOURCES = 10; + + /** + * Virtual threads require Java 21, on an older JVM the framework transparently falls back to + * platform threads, which this test also covers since only the concurrency assertions apply then. + */ + private static final boolean VIRTUAL_THREADS_SUPPORTED = Runtime.version().feature() >= 21; + + @RegisterExtension + LocallyRunOperatorExtension operator = + LocallyRunOperatorExtension.builder() + .withConfigurationService( + o -> + o.withUseVirtualThreads(true) + .withConcurrentReconciliationThreads(CONCURRENT_RECONCILIATION_THREADS)) + .withReconciler(new VirtualThreadsTestReconciler()) + .build(); + + @Test + void reconciliationsRunOnVirtualThreadsWithinTheConfiguredConcurrency() { + // the test itself runs on a platform thread, so the reconciler assertion below can only pass + // if the framework actually switched to virtual threads + assertThat(onVirtualThread()).isFalse(); + + IntStream.range(0, NUMBER_OF_RESOURCES).forEach(i -> operator.create(testResource(i))); + + var reconciler = operator.getReconcilerOfType(VirtualThreadsTestReconciler.class); + await() + .atMost(2, TimeUnit.MINUTES) + .untilAsserted( + () -> + assertThat(reconciler.getNumberOfExecutions()) + .isGreaterThanOrEqualTo(NUMBER_OF_RESOURCES)); + + // parallelism is retained: reconciliations do happen concurrently, but never more than the + // configured number of them + assertThat(reconciler.getMaxConcurrentReconciliations()) + .isEqualTo(CONCURRENT_RECONCILIATION_THREADS); + assertThat(reconciler.allExecutionsOnVirtualThreads()).isEqualTo(VIRTUAL_THREADS_SUPPORTED); + } + + private VirtualThreadsCustomResource testResource(int index) { + var resource = new VirtualThreadsCustomResource(); + resource.setMetadata(new ObjectMeta()); + resource.getMetadata().setName("virtual-threads-test-" + index); + return resource; + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsTestReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsTestReconciler.java new file mode 100644 index 0000000000..ee6f00f499 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsTestReconciler.java @@ -0,0 +1,84 @@ +/* + * 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.baseapi.virtualthreads; + +import java.time.Duration; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import io.javaoperatorsdk.operator.api.reconciler.Context; +import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.Reconciler; +import io.javaoperatorsdk.operator.api.reconciler.UpdateControl; +import io.javaoperatorsdk.operator.support.TestExecutionInfoProvider; + +/** + * Blocks for a while during reconciliation, recording on what kind of thread it ran and how many + * reconciliations were in flight at the same time. + */ +@ControllerConfiguration +public class VirtualThreadsTestReconciler + implements Reconciler, TestExecutionInfoProvider { + + public static final Duration RECONCILIATION_DURATION = Duration.ofMillis(300); + + private final AtomicInteger numberOfExecutions = new AtomicInteger(); + private final AtomicInteger runningReconciliations = new AtomicInteger(); + private final AtomicInteger maxConcurrentReconciliations = new AtomicInteger(); + private final AtomicBoolean allExecutionsOnVirtualThreads = new AtomicBoolean(true); + + @Override + public UpdateControl reconcile( + VirtualThreadsCustomResource resource, Context context) + throws InterruptedException { + if (!onVirtualThread()) { + allExecutionsOnVirtualThreads.set(false); + } + maxConcurrentReconciliations.accumulateAndGet( + runningReconciliations.incrementAndGet(), Math::max); + try { + Thread.sleep(RECONCILIATION_DURATION.toMillis()); + } finally { + runningReconciliations.decrementAndGet(); + numberOfExecutions.incrementAndGet(); + } + return UpdateControl.noUpdate(); + } + + public int getNumberOfExecutions() { + return numberOfExecutions.get(); + } + + public int getMaxConcurrentReconciliations() { + return maxConcurrentReconciliations.get(); + } + + public boolean allExecutionsOnVirtualThreads() { + return allExecutionsOnVirtualThreads.get(); + } + + /** + * {@code Thread.isVirtual} only exists as of Java 21 while the tests are compiled for Java 17, + * hence the reflective call. + */ + static boolean onVirtualThread() { + try { + return (Boolean) Thread.class.getMethod("isVirtual").invoke(Thread.currentThread()); + } catch (ReflectiveOperationException e) { + return false; + } + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/config/loader/ConfigLoaderTest.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/config/loader/ConfigLoaderTest.java index 3ac9cea0dc..e604cdac35 100644 --- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/config/loader/ConfigLoaderTest.java +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/config/loader/ConfigLoaderTest.java @@ -94,6 +94,7 @@ void applyConfigsAppliesBooleanFlags() { values.put("josdk.dependent-resources.ssa-based-create-update-match", false); values.put("josdk.use-ssa-to-patch-primary-resource", false); values.put("josdk.clone-secondary-resources-when-getting-from-cache", true); + values.put("josdk.use-virtual-threads", true); var loader = new ConfigLoader(mapProvider(values)); var base = new BaseConfigurationService(null); @@ -106,6 +107,7 @@ void applyConfigsAppliesBooleanFlags() { assertThat(result.ssaBasedCreateUpdateMatchForDependentResources()).isFalse(); assertThat(result.useSSAToPatchPrimaryResource()).isFalse(); assertThat(result.cloneSecondaryResourcesWhenGettingFromCache()).isTrue(); + assertThat(result.useVirtualThreads()).isTrue(); } @Test From 6826eb9cbd2a8c74cf4631cbb571f9ac10390a80 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Attila=20M=C3=A9sz=C3=A1ros?= Date: Thu, 17 Sep 2026 08:50:05 +0200 Subject: [PATCH 2/6] wip MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Attila Mészáros --- .../operator/api/config/ExecutorServiceManager.java | 2 ++ .../operator/api/config/VirtualThreads.java | 8 ++++++++ 2 files changed, 10 insertions(+) diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManager.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManager.java index 2176cb0fab..342e07fb9f 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManager.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManager.java @@ -168,6 +168,8 @@ public synchronized void start(ConfigurationService configurationService) { this.configurationService = configurationService; // used to lazy init workflow executor 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; diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java index dd7a8a80a0..d80d3928b8 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java @@ -39,6 +39,14 @@ *

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. + * + *

There is intentionally no scheduled variant here: the JDK provides no virtual thread backed + * {@link java.util.concurrent.ScheduledExecutorService}, scheduling was deliberately left out of + * virtual threads (the Loom runtime itself uses a platform thread scheduler to unpark timed out + * virtual threads). Running scheduled tasks on virtual threads would mean keeping a platform thread + * scheduler purely for the timing and handing each fired task off to a virtual thread executor, + * which the SDK doesn't do, so {@link ExecutorServiceManager#scheduledExecutorService()} is always + * backed by platform threads. */ final class VirtualThreads { From 24701dc4d38baea75898a4e5628863cbe4082680 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Attila=20M=C3=A9sz=C3=A1ros?= Date: Thu, 17 Sep 2026 09:20:50 +0200 Subject: [PATCH 3/6] wip MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Attila Mészáros --- .../operator/api/config/ConfigurationService.java | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java index e3454e6c98..d7cfc049cf 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java @@ -202,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 * + *

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() { @@ -212,6 +217,12 @@ default int concurrentReconciliationThreads() { * Number of threads the operator can spin out to be used in the workflows with the default * executor. * + *

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() { From 1f6760e5779a58df5d8dcd355bcf17ae342eb75a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Attila=20M=C3=A9sz=C3=A1ros?= Date: Fri, 18 Sep 2026 14:45:45 +0200 Subject: [PATCH 4/6] remove unnecessary comment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Attila Mészáros --- .../io/javaoperatorsdk/operator/api/config/VirtualThreads.java | 1 - 1 file changed, 1 deletion(-) diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java index d80d3928b8..1d84c82e25 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java @@ -131,7 +131,6 @@ private static final class BoundedExecutorService extends AbstractExecutorServic 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); } From 3d177f0868988827fdeefea97a0383eade43883e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Attila=20M=C3=A9sz=C3=A1ros?= Date: Mon, 21 Sep 2026 13:53:05 +0200 Subject: [PATCH 5/6] wip MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Attila Mészáros --- .../en/docs/documentation/operations/configuration.md | 8 ++++++-- .../operator/api/config/ConfigurationService.java | 8 +++++--- .../api/config/ConfigurationServiceOverrider.java | 2 +- .../operator/api/config/VirtualThreads.java | 6 ++++-- .../javaoperatorsdk/operator/sample/WebPageOperator.java | 4 +++- 5 files changed, 19 insertions(+), 9 deletions(-) diff --git a/docs/content/en/docs/documentation/operations/configuration.md b/docs/content/en/docs/documentation/operations/configuration.md index b5a71bb126..2337ae420d 100644 --- a/docs/content/en/docs/documentation/operations/configuration.md +++ b/docs/content/en/docs/documentation/operations/configuration.md @@ -45,7 +45,11 @@ 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 +- 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 @@ -293,7 +297,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-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 | diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java index d7cfc049cf..a2a67ad0a2 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java @@ -251,9 +251,11 @@ default Metrics getMetrics() { * 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. * - *

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. + *

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 JEP 491 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. * *

Note that this only affects the executors created by the framework: a custom {@link * ExecutorService} provided through {@link #getExecutorService()} or {@link diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java index b3ae079561..0f4f74bf37 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java @@ -122,7 +122,7 @@ public ConfigurationServiceOverrider withWorkflowExecutorService( /** * 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 + * platform threads. Officially supported on Java 25 or later, see {@link * ConfigurationService#useVirtualThreads()} for the details. * * @param useVirtualThreads {@code true} to use virtual threads diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java index 1d84c82e25..ae76823fbc 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java @@ -89,7 +89,8 @@ static boolean shouldUse(boolean 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.", + + " or later is required, Java 25 or later is officially supported). Falling back to" + + " platform threads.", Runtime.version().feature()); } return false; @@ -99,7 +100,8 @@ static boolean shouldUse(boolean requested) { static ExecutorService newVirtualThreadPerTaskExecutor() { if (!isSupported()) { throw new OperatorException( - "Virtual threads are not supported by the JVM in use, Java 21 or later is required"); + "Virtual threads are not supported by the JVM in use, Java 21 or later is required" + + " (Java 25 or later is officially supported)"); } try { return (ExecutorService) NEW_VIRTUAL_THREAD_PER_TASK_EXECUTOR.invokeExact(); diff --git a/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageOperator.java b/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageOperator.java index 5366dc2e9a..392172886f 100644 --- a/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageOperator.java +++ b/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageOperator.java @@ -40,7 +40,9 @@ public class WebPageOperator { public static void main(String[] args) throws IOException { log.info("WebServer Operator starting!"); - Operator operator = new Operator(o -> o.withStopOnInformerErrorDuringStartup(false)); + Operator operator = + new Operator( + o -> o.withStopOnInformerErrorDuringStartup(false).withUseVirtualThreads(true)); String reconcilerEnvVar = System.getenv(WEBPAGE_RECONCILER_ENV); if (WEBPAGE_CLASSIC_RECONCILER_ENV_VALUE.equals(reconcilerEnvVar)) { operator.register(new WebPageReconciler()); From b87ca8742934aa44162e3e8543726b171b747c5f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Attila=20M=C3=A9sz=C3=A1ros?= Date: Wed, 23 Sep 2026 13:33:51 +0200 Subject: [PATCH 6/6] feat: run default Kubernetes client tasks on virtual threads When useVirtualThreads() is enabled, the client created by the default ConfigurationService#getKubernetesClient() now uses a virtual thread task executor (informer event dispatching, watch event delivery) instead of fabric8's cached platform thread pool, shut down when the client is closed. User provided clients are used as is. --- .../documentation/operations/configuration.md | 9 +++++ .../api/config/ConfigurationService.java | 32 ++++++++++++----- .../operator/api/config/VirtualThreads.java | 22 ++++++++++++ .../api/config/VirtualThreadsTest.java | 35 +++++++++++++++++++ 4 files changed, 90 insertions(+), 8 deletions(-) diff --git a/docs/content/en/docs/documentation/operations/configuration.md b/docs/content/en/docs/documentation/operations/configuration.md index 2337ae420d..20b3ca2003 100644 --- a/docs/content/en/docs/documentation/operations/configuration.md +++ b/docs/content/en/docs/documentation/operations/configuration.md @@ -54,6 +54,15 @@ Two things to keep in mind: 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 diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java index a2a67ad0a2..1d750bdc96 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java @@ -154,6 +154,12 @@ public R clone(R object) { * io.javaoperatorsdk.operator.Operator#Operator(Consumer)}, passing your custom instance with * {@link ConfigurationServiceOverrider#withKubernetesClient(KubernetesClient)}. * + *

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. + * *

NOTE: 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. @@ -162,13 +168,17 @@ public 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(); } /** @@ -259,7 +269,13 @@ default Metrics getMetrics() { * *

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. + * #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 diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java index ae76823fbc..7c7b8edc2b 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java @@ -20,6 +20,7 @@ import java.lang.invoke.MethodType; import java.util.List; import java.util.concurrent.AbstractExecutorService; +import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; @@ -30,6 +31,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import io.fabric8.kubernetes.client.KubernetesClientBuilder.ExecutorSupplier; import io.javaoperatorsdk.operator.OperatorException; /** @@ -110,6 +112,26 @@ static ExecutorService newVirtualThreadPerTaskExecutor() { } } + /** + * Supplies the task executor of the default {@link io.fabric8.kubernetes.client.KubernetesClient} + * created by {@link ConfigurationService#getKubernetesClient()}: an unbounded virtual thread + * executor, replacing the client's default cached platform thread pool, that is shut down when + * the client is closed. + */ + static ExecutorSupplier newKubernetesClientTaskExecutorSupplier() { + return new ExecutorSupplier() { + @Override + public Executor get() { + return newVirtualThreadPerTaskExecutor(); + } + + @Override + public void onClose(Executor executor) { + ((ExecutorService) executor).shutdownNow(); + } + }; + } + /** * A virtual thread based executor executing at most {@code maxConcurrency} tasks at the same * time, the equivalent of a fixed size platform thread pool. diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VirtualThreadsTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VirtualThreadsTest.java index 60d7e26f54..c28be4b4fc 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VirtualThreadsTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VirtualThreadsTest.java @@ -16,6 +16,7 @@ package io.javaoperatorsdk.operator.api.config; import java.util.Set; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; @@ -28,6 +29,9 @@ import org.junit.jupiter.api.condition.EnabledForJreRange; import org.junit.jupiter.api.condition.JRE; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.impl.BaseClient; + import static org.assertj.core.api.Assertions.assertThat; class VirtualThreadsTest { @@ -148,6 +152,37 @@ void shutdownTerminatesOnceTheAlreadySubmittedTasksAreDone() throws Exception { assertThat(done.getCount()).isZero(); } + @Test + void defaultKubernetesClientKeepsPlatformThreadsWhenVirtualThreadsAreNotRequested() + throws Exception { + try (var client = defaultKubernetesClient(false)) { + assertThat(runsTasksOnVirtualThread(client)).isFalse(); + } + } + + @Test + @EnabledForJreRange(min = JRE.JAVA_21) + void defaultKubernetesClientRunsItsTasksOnVirtualThreadsWhenRequested() throws Exception { + ExecutorService executor; + try (var client = defaultKubernetesClient(true)) { + assertThat(runsTasksOnVirtualThread(client)).isTrue(); + executor = (ExecutorService) client.adapt(BaseClient.class).getExecutor(); + } + assertThat(executor.isShutdown()).isTrue(); + } + + private static KubernetesClient defaultKubernetesClient(boolean useVirtualThreads) { + return ConfigurationService.newOverriddenConfigurationService( + o -> o.withUseVirtualThreads(useVirtualThreads)) + .getKubernetesClient(); + } + + private static boolean runsTasksOnVirtualThread(KubernetesClient client) throws Exception { + return CompletableFuture.supplyAsync( + VirtualThreadsTest::onVirtualThread, client.adapt(BaseClient.class).getExecutor()) + .get(TIMEOUT_SECONDS, TimeUnit.SECONDS); + } + /** * {@code Thread.isVirtual} only exists as of Java 21 while the tests are compiled for Java 17, * hence the reflective call.