Skip to content

Commit 07fb4e3

Browse files
csviriclaude
andauthored
feat: virtual thread support (#3664)
Add the useVirtualThreads flag (ConfigurationServiceOverrider. withUseVirtualThreads, josdk.use-virtual-threads) that runs the framework's concurrent work on virtual threads. It is off by default and only supported on Java 25 or later (JEP 491); on an older JVM a warning is logged and platform threads are used. With the flag on: - reconciliations and dependent workflows run on virtual threads, still capped by concurrentReconciliationThreads and concurrentWorkflowExecutorThreads - the caching executor used for internal housekeeping uses a virtual thread per task - the Kubernetes client created by the framework runs its internal task executor on virtual threads, a provided client is used as is - scheduled tasks run on virtual threads through the new VirtualThreadScheduledExecutor, a daemon timing thread firing each task on a virtual thread, with the semantics of the platform default (waiting and periodic tasks discarded on shutdown, running tasks let finish, fixed rate tasks catch up after an overrun) Scheduled work no longer relies on java.util.Timer: - PollingEventSource and PerResourcePollingEventSource poll on an executor of their own, provided through the configuration builder or created, and shut down, by the event source, on virtual threads when the flag is on. A provided executor is never shut down by the source. They can be started again after a stop, and a poll in flight is not interrupted by it. - PerResourcePollingEventSource registers the resources of the primary cache when it starts, drops the result of a poll of a previous run, and forgets the registration of a resource that is gone from the cache so that it is polled again if recreated - TimerEventSource schedules on the retry and reschedule executor managed by ExecutorServiceManager, see ConfigurationService.getRetryAndRescheduleExecutorService and concurrentRetryAndRescheduleThreads. Timers pending on an executor the operator replaced are not scheduled again. - the default pools use daemon threads, discard delayed tasks on shutdown and let idle threads time out Behavior and API changes, described in the v5.7 migration guide: - an executor provided to PerResourcePollingConfigurationBuilder is not shut down by the event source anymore, and the configuration returns a null executor when none was provided - PollingEventSource(Class, PollingConfiguration), the TimerEventSource(String, boolean) constructor and ExecutorServiceManager.scheduledExecutorService() are deprecated - ConfigurationServiceOverrider rebuilds the default retry executor when the flag or its thread count changes, a custom executor of the wrapped configuration is then replaced, see the Javadoc Also: configuration keys in ConfigLoader, documentation (Virtual Threads, polling event source threading, migration guide), the webpage and mysql-schema samples run with virtual threads, VirtualThreadsIT, and a spotbugs exclusion for the scheduled executor handles. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
1 parent 6558161 commit 07fb4e3

41 files changed

Lines changed: 3411 additions & 101 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

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

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -222,6 +222,37 @@ is similar to `PerResourcePollingEventSource` except that, contrary to that even
222222
doesn't poll a specific API separately per resource, but periodically and independently of
223223
actually observed primary resources.
224224

225+
#### Threading of the polling event sources
226+
227+
Each polling event source polls on a scheduled executor of its own, so a slow poll of one event
228+
source never delays the polls of the others. Unless you provide one, the event source creates a
229+
single daemon thread executor when it is started and shuts it down when it is stopped. With a single
230+
thread the polls of a source run one after the other. A `PollingEventSource` runs a single periodic
231+
task, so more threads would not help it. A `PerResourcePollingEventSource` however polls each
232+
registered resource on its own, so if it has many registered resources or fetches slowly, it can be
233+
given an executor with more threads using the `withExecutorService` method of its configuration
234+
builder:
235+
236+
```java
237+
var config =
238+
new PerResourcePollingConfigurationBuilder<>(this::fetchResources, Duration.ofSeconds(5))
239+
.withExecutorService(Executors.newScheduledThreadPool(4))
240+
.build();
241+
```
242+
243+
An executor provided that way is not managed by the operator: it is your responsibility to shut it
244+
down. It is only ever used as is, the virtual threads flag has no effect on it. You can also share
245+
it between multiple event sources deliberately.
246+
247+
With virtual threads enabled, see [virtual threads](../operations/configuration/#virtual-threads),
248+
the executor the event source creates runs the polls on virtual threads, only keeping one platform
249+
thread to do the timing. The number of polls running at the same time is then not limited.
250+
251+
Retried and rescheduled reconciliations are triggered on a separate executor, so a slow poll can
252+
never delay them. It is sized with `withConcurrentRetryAndRescheduleThreads` (2 threads by
253+
default); few threads are needed there since triggering a reconciliation only enqueues an event
254+
for one of the reconciliation threads to pick up.
255+
225256
#### Inbound event sources
226257

227258
[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‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,60 @@ Operator operator = new Operator( override -> override
2323
.withLeaderElectionConfiguration(new LeaderElectionConfiguration("bar", "barNS")));
2424
```
2525

26+
### Virtual Threads
27+
28+
Reconciliation is mostly about blocking: talking to the Kubernetes API server or to external
29+
systems. Virtual threads make such blocking calls much cheaper than platform threads, and the
30+
framework can be switched over to them with a single flag:
31+
32+
```java
33+
Operator operator = new Operator(override -> override.withUseVirtualThreads(true));
34+
```
35+
36+
When enabled, reconciliations, dependent resource workflows and the framework's internal
37+
housekeeping (starting the informers, for example) all run on virtual threads.
38+
39+
Enabling virtual threads does **not** remove the concurrency limits, parallelism is configured
40+
exactly as before: `withConcurrentReconciliationThreads(int)` still caps how many reconciliations
41+
run at the same time and `withConcurrentWorkflowExecutorThreads(int)` how many dependent resources
42+
of a workflow are processed concurrently. Only the threads backing those limits change. Since
43+
virtual threads are cheap, these limits can usually be raised significantly compared to what is
44+
reasonable with platform threads.
45+
46+
Things to keep in mind:
47+
48+
- Virtual threads are only supported on **Java 25 or later**. Before Java 25 a virtual thread pins
49+
its carrier thread while it is inside a `synchronized` block, which can starve the carrier pool
50+
and even deadlock the operator on startup. [JEP 491](https://openjdk.org/jeps/491), delivered in
51+
Java 25, removed that pinning. This is checked at runtime: on an older JVM, including the ones
52+
where virtual threads exist (Java 21 to 24), a warning is logged and platform threads are used
53+
instead, so the same configuration works on any Java version the operator runs on.
54+
- The scheduled tasks (the polls of the polling event sources, which each create an executor of
55+
their own unless one is provided, and the retried and rescheduled reconciliations) run on virtual
56+
threads too. The JDK has no scheduled executor backed by virtual threads, so the default ones keep
57+
a single platform thread for the timing only and run each task they fire on a virtual thread. The
58+
number of those tasks running at the same time is not limited, so
59+
`withConcurrentRetryAndRescheduleThreads(int)` only applies to platform threads.
60+
- Virtual threads are always daemon threads, so unlike the threads of the platform thread pools they
61+
don't keep the JVM running: if the `main` method of your application returns after starting the
62+
operator and nothing else keeps the JVM alive, call `Operator.keepAlive()`, see
63+
[Keeping the JVM Alive](#keeping-the-jvm-alive).
64+
- A custom `ExecutorService` provided through `withExecutorService(...)` or
65+
`withWorkflowExecutorService(...)` is always used as is, the flag has no effect on it. The same
66+
goes for `withRetryAndRescheduleExecutorService(...)`. Note however that overriding the flag, or
67+
`withConcurrentRetryAndRescheduleThreads(int)`, of a configuration that provides its own retry and
68+
reschedule executor replaces that executor with the default one, so provide it again with
69+
`withRetryAndRescheduleExecutorService(...)` in that case.
70+
- The Kubernetes client the framework creates when none is provided also switches its internal
71+
task executor (used to dispatch informer events to their handlers and to deliver watch events)
72+
to virtual threads. A client you provide through `withKubernetesClient(...)` is used as is: to
73+
get the same behavior, configure it yourself, e.g. with
74+
`new KubernetesClientBuilder().withTaskExecutor(Executors.newVirtualThreadPerTaskExecutor())`
75+
(the client doesn't shut down an executor passed that way, use `withTaskExecutorSupplier(...)`
76+
if it should be shut down when the client is closed).
77+
Either way, the blocking calls your reconciler makes through the client work well on virtual
78+
threads without any client-side change.
79+
2680
## Reconciler-Level Configuration
2781

2882
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.`.
285339
|---|---|---|
286340
| `josdk.check-crd` | `Boolean` | Validate CRDs against local model on startup |
287341
| `josdk.close-client-on-stop` | `Boolean` | Close the Kubernetes client when the operator stops |
342+
| `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) |
288343
| `josdk.use-ssa-to-patch-primary-resource` | `Boolean` | Use Server-Side Apply to patch the primary resource |
289344
| `josdk.clone-secondary-resources-when-getting-from-cache` | `Boolean` | Clone secondary resources on cache reads |
290345

@@ -301,6 +356,12 @@ All operator-level keys are prefixed with `josdk.`.
301356
|---|---|---|
302357
| `josdk.workflow.executor-threads` | `Integer` | Thread pool size for workflow execution |
303358

359+
#### Retry and Reschedule
360+
361+
| Key | Type | Description |
362+
|---|---|---|
363+
| `josdk.retry-and-reschedule.concurrent-threads` | `Integer` | Thread pool size for triggering retried and rescheduled reconciliations |
364+
304365
#### Informer
305366

306367
| Key | Type | Description |

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

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -78,3 +78,35 @@ on the classpath next to your implementation. This is especially important if yo
7878
based client (`kubernetes-httpclient-vertx`), since Vert.x 4 and Vert.x 5 cannot be used together.
7979
If you want to keep the Vert.x 4 based client, use it in place of `kubernetes-httpclient-jdk` in the
8080
snippet above.
81+
82+
## Executors of the polling and timer event sources
83+
84+
The polling event sources and the internal `TimerEventSource` no longer use a `java.util.Timer`, and
85+
`PerResourcePollingEventSource` no longer creates its executor when its configuration is built. They
86+
run on scheduled executors, see [virtual threads](../documentation/operations/configuration/#virtual-threads)
87+
and [Threading of the polling event sources](../documentation/eventing/#threading-of-the-polling-event-sources).
88+
This is transparent unless your operator provides an executor to a polling event source, or relies
89+
on one of the following:
90+
91+
- A `ScheduledExecutorService` provided with `withExecutorService(...)` to a
92+
`PerResourcePollingConfigurationBuilder` was shut down with `shutdownNow()` when the event source
93+
was stopped. It is not anymore: **an executor you provide is yours to shut down**, like it is for
94+
`PollingConfigurationBuilder`. If you relied on the event source for this, shut it down yourself
95+
after stopping the operator, otherwise its threads can keep the JVM from exiting if they are not
96+
daemon threads.
97+
- `PerResourcePollingConfiguration.executorService()` returns `null` when no executor was provided,
98+
it used to be a `ScheduledThreadPoolExecutor` with a single non-daemon thread created by the
99+
configuration. The event source now creates a single daemon thread executor, or one running on
100+
virtual threads if `useVirtualThreads` is enabled, when it starts, shuts it down when it stops, and
101+
creates a new one if it is started again. `PerResourcePollingConfiguration.DEFAULT_EXECUTOR_THREAD_NUMBER`
102+
is deprecated, it isn't used anymore.
103+
- Each polling event source has an executor of its own: they don't share threads, and the number of
104+
threads of a source is not configurable other than by providing an executor.
105+
- `new PollingEventSource<>(Class, PollingConfiguration)` is deprecated. Use the constructor that
106+
takes the `EventSourceContext` as well, `new PollingEventSource<>(Class, EventSourceContext,
107+
PollingConfiguration)`, otherwise the event source cannot know about the `useVirtualThreads`
108+
setting and polls on a platform thread.
109+
- `TimerEventSource(String, boolean)` is deprecated, use the constructor that also takes an
110+
executor supplier, `null` to have the event source create an executor of its own.
111+
- `ExecutorServiceManager.scheduledExecutorService()` is deprecated. The framework doesn't use it
112+
anymore, it creates the executor on first use, and shuts it down when the manager is stopped.
Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,77 @@
1+
/*
2+
* Copyright Java Operator SDK Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package io.javaoperatorsdk.operator;
17+
18+
import java.util.concurrent.Executors;
19+
import java.util.concurrent.ScheduledExecutorService;
20+
import java.util.concurrent.ScheduledThreadPoolExecutor;
21+
import java.util.concurrent.ThreadFactory;
22+
import java.util.concurrent.TimeUnit;
23+
import java.util.concurrent.atomic.AtomicLong;
24+
25+
public class ExecutorUtilsInternal {
26+
27+
private static final long SCHEDULED_THREAD_KEEP_ALIVE_SECONDS = 60;
28+
29+
// prevent instantiation of util class
30+
private ExecutorUtilsInternal() {}
31+
32+
/**
33+
* Creates a {@link ThreadFactory} producing daemon threads named after the specified prefix.
34+
* Daemon threads don't keep the JVM alive if the {@link Operator} is never stopped, and naming
35+
* them makes the pool they belong to identifiable in thread dumps.
36+
*
37+
* @param namePrefix the prefix the created threads are named after
38+
* @return a {@link ThreadFactory} creating named daemon threads
39+
*/
40+
public static ThreadFactory daemonThreadFactory(String namePrefix) {
41+
final var defaultThreadFactory = Executors.defaultThreadFactory();
42+
final var counter = new AtomicLong();
43+
return runnable -> {
44+
final var thread = defaultThreadFactory.newThread(runnable);
45+
thread.setName(namePrefix + "-" + counter.incrementAndGet());
46+
thread.setDaemon(true);
47+
return thread;
48+
};
49+
}
50+
51+
/**
52+
* Creates the kind of {@link ScheduledExecutorService} the operator runs its scheduled (i.e.
53+
* periodic or delayed) tasks on: one whose threads are daemon threads named after the specified
54+
* prefix, and which doesn't let the tasks that were scheduled for later delay its shutdown.
55+
*
56+
* @param corePoolSize the maximal number of threads of the pool, which are created on demand and
57+
* let go if idle for a while
58+
* @param threadNamePrefix the prefix the threads of the pool are named after
59+
* @return a {@link ScheduledExecutorService} to run scheduled tasks on
60+
*/
61+
public static ScheduledExecutorService daemonScheduledThreadPool(
62+
int corePoolSize, String threadNamePrefix) {
63+
final var executor =
64+
new ScheduledThreadPoolExecutor(corePoolSize, daemonThreadFactory(threadNamePrefix));
65+
// tasks that are scheduled far out (a reconciliation rescheduled in an hour, say) would
66+
// otherwise keep the pool from terminating until the graceful shutdown timeout expires
67+
executor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
68+
// cancelled tasks are frequent (every retry that is superseded by a new event cancels one) and
69+
// would otherwise be retained until their delay elapses
70+
executor.setRemoveOnCancelPolicy(true);
71+
// the threads are created as tasks get scheduled, not as the load requires, so without this
72+
// they would all stay around idle for good; the pool always keeps a thread for delayed tasks
73+
executor.setKeepAliveTime(SCHEDULED_THREAD_KEEP_ALIVE_SECONDS, TimeUnit.SECONDS);
74+
executor.allowCoreThreadTimeOut(true);
75+
return executor;
76+
}
77+
}

0 commit comments

Comments
 (0)