Repository navigation
Conversation
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the
|
There was a problem hiding this comment.
🟡 Changes recommended
Unresolved executor lifecycle, polling startup, and event-validation issues can lose work or prevent events from being recorded.
4 open issues
What changed in this PR
Adds opt-in virtual threads to Java Operator SDK while preserving concurrency limits and Java 17 fallback. The PR also centralizes scheduling and includes event-recording and desired-state fixes.
Changes:
- Adds virtual-thread configuration, executor implementations, client integration, documentation, and tests.
- Moves polling and retry scheduling to separate configurable executors with lifecycle management.
- Updates event aggregation, long event names, and desired-state aspect handling.
| File | Description |
|---|---|
| sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageOperator.java | Enables virtual threads in the sample. |
| operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/multiplemanagedexternaldependenttype/MultipleManagedExternalDependentResourceReconciler.java | Supplies context for shared polling. |
| operator-framework/src/test/java/io/javaoperatorsdk/operator/config/loader/ConfigLoaderTest.java | Tests new configuration keys. |
| operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsTestReconciler.java | Records thread type and concurrency. |
| operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsIT.java | Tests reconciliation threading and limits. |
| operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsCustomResource.java | Provides the integration-test resource. |
| operator-framework/src/main/java/io/javaoperatorsdk/operator/config/loader/ConfigLoader.java | Binds virtual-thread and scheduling settings. |
| operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/timer/TimerEventSourceTest.java | Tests executor ownership and restart. |
| operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSourceTest.java | Tests shared scheduling and cancellation. |
| operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSourceTest.java | Tests polling restart behavior. |
| operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/DesiredStateAspectAuditTest.java | Adds desired-state aspect regression cases. |
| operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/event/EventRecorderAuditTest.java | Adds event-recording regression cases. |
| operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/event/DefaultEventRecorderTest.java | Tests collision-resistant name shortening. |
| operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VirtualThreadsTest.java | Tests executor and client threading. |
| operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManagerTest.java | Tests scheduler separation and lifecycle. |
| operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverriderTest.java | Tests threading configuration overrides. |
| operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/BoundedVirtualThreadExecutorAuditTest.java | Tests limits and pending-task shutdown. |
| operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/timer/TimerEventSource.java | Replaces timers with supplied or owned executors. |
| operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSource.java | Adds shared scheduling and executor ownership. |
| operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingConfigurationBuilder.java | Adds custom polling executor configuration. |
| operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingConfiguration.java | Stores the optional polling executor. |
| operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSource.java | Resolves executors on startup and cancels tasks. |
| operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingConfiguration.java | Defers default executor selection. |
| operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSources.java | Accepts a retry scheduler supplier. |
| operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSourceManager.java | Connects retries to the managed scheduler. |
| operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/PollingDependentResource.java | Passes context to polling sources. |
| operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/BulkDependentResourceReconciler.java | Passes the user dependent to aspects. |
| operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContext.java | Handles reentrant desired-state aspect lookups. |
| operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/event/DefaultEventSink.java | Updates fields during named-event aggregation. |
| operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/event/DefaultEventRecorder.java | Adds event timestamps and hashed name suffixes. |
| operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java | Implements virtual executors and concurrency bounds. |
| operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/Utils.java | Adds named daemon scheduler utilities. |
| operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManager.java | Selects executors and manages scheduler lifecycle. |
| operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java | Exposes threading and scheduler overrides. |
| operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java | Defines threading defaults and client integration. |
| docs/content/en/docs/documentation/operations/configuration.md | Documents virtual threads and configuration keys. |
| docs/content/en/docs/documentation/eventing.md | Documents shared polling and executor ownership. |
Review effort: Balanced
There was a problem hiding this comment.
🟡 Changes recommended
Unresolved scheduler contract violations and cancellation/lifecycle races can skip fixed-rate executions, run cancelled work, and leave duplicate or stale polling chains.
7 open issues
Return unstarted scheduled work from shutdownNow() · New Remove cancelled handles from waiting during rearm · New Publish runner before cancellation check · New Preserve overdue deadlines for fixed-rate task catch-up · New Prevent stale replay registrations after deletion · New Invalidate stale polling chains across lifecycle restarts · New Wait for event-producing thread names before consuming them · New
4 resolved since last review
Review effort: Balanced
There was a problem hiding this comment.
🟡 Changes recommended
Unresolved lifecycle races can resurrect or duplicate polling tasks, while premature non-daemon thread startup can prevent process exit after initialization failure.
14 open issues
Operator.init()obtains this manager, whose constructor callsstart(), so merely constructing… · New A deletion between removing a buffered registration and installing its future finds nothing to… · New If an in-flight fetch returns afterstop()and a subsequentstart(), this check sees the source… · NewshutdownNow()omits scheduled tasks that never started: it discards both waiting timers and… · New Cancellation can occur after theisDone()check but beforerunneris assigned, so… · New After a fixed-rate task overruns its period,rearm()clamps the delay to zero andarm()resets… · New The event count does not guarantee that the thread-name entry is available.CapturingEventHandler… · New Wait for event-producing thread names before consuming them Invalidate stale polling chains across lifecycle restarts Prevent stale replay registrations after deletion Preserve overdue deadlines for fixed-rate task catch-up Publish runner before cancellation check Remove cancelled handles from waiting during rearm Return unstarted scheduled work from shutdownNow()
Review effort: Balanced
There was a problem hiding this comment.
🔵 Needs a closer look
Critical polling lifecycle races can schedule deleted resources, prevent polling after recreation, and allow stale fetches to create duplicate polling chains after restart.
14 open issues
If an in-flight fetch returns afterstop()and a subsequentstart(), this check sees the source… A deletion between removing a buffered registration and installing its future finds nothing to…Operator.init()obtains this manager, whose constructor callsstart(), so merely constructing… The event count does not guarantee that the thread-name entry is available.CapturingEventHandler… After a fixed-rate task overruns its period,rearm()clamps the delay to zero andarm()resets… Cancellation can occur after theisDone()check but beforerunneris assigned, so…shutdownNow()omits scheduled tasks that never started: it discards both waiting timers and… Wait for event-producing thread names before consuming them Invalidate stale polling chains across lifecycle restarts Prevent stale replay registrations after deletion Preserve overdue deadlines for fixed-rate task catch-up Publish runner before cancellation check Remove cancelled handles from waiting during rearm Return unstarted scheduled work from shutdownNow()
Review effort: Balanced
There was a problem hiding this comment.
🟡 Changes recommended
Operator construction prematurely starts a non-daemon keep-alive thread, preventing JVM exit if the operator is never started or reconciler registration fails.
7 open issues
If an in-flight fetch returns afterstop()and a subsequentstart(), this check sees the source…Operator.init()obtains this manager, whose constructor callsstart(), so merely constructing… Virtual-thread flag overrides inherited custom executors · NewshutdownNow()omits scheduled tasks that never started: it discards both waiting timers and… Invalidate stale polling chains across lifecycle restarts Remove cancelled handles from waiting during rearm Return unstarted scheduled work from shutdownNow()
8 resolved since last review
A deletion between removing a buffered registration and installing its future finds nothing to… The event count does not guarantee that the thread-name entry is available.CapturingEventHandler… After a fixed-rate task overruns its period,rearm()clamps the delay to zero andarm()resets… Cancellation can occur after theisDone()check but beforerunneris assigned, so… Wait for event-producing thread names before consuming them Prevent stale replay registrations after deletion Preserve overdue deadlines for fixed-rate task catch-up Publish runner before cancellation check
Review effort: Balanced
There was a problem hiding this comment.
🟡 Changes recommended
Unresolved polling lifecycle races can schedule deleted resources, prevent polling after resource recreation, and create duplicate polling chains after restart.
7 open issues
If an in-flight fetch returns afterstop()and a subsequentstart(), this check sees the source… Add lifecycle-managed non-daemon keep-alive thread · New Virtual-thread flag overrides inherited custom executorsshutdownNow()omits scheduled tasks that never started: it discards both waiting timers and… Invalidate stale polling chains across lifecycle restarts Remove cancelled handles from waiting during rearm Return unstarted scheduled work from shutdownNow()
1 resolved since last review
Review effort: Balanced
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Publish the runner of a scheduled task before checking whether it was cancelled, make fixed-rate tasks catch up after an overrun, drop the stale scheduled future of a primary missing from the cache, and fix a race in TimerEventSourceTest.
Signed-off-by: Attila Mészáros <a_meszaros@apple.com>
ScheduledThreadPoolExecutor starts a core thread on every schedule call until the core size is reached, regardless of the load, and never gives them up. Allow the core threads to time out so that the pools shrink back when idle.
…ranch Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Signed-off-by: Attila Mészáros <a_meszaros@apple.com>
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
…d handles from the waiting set - a per resource poll that outlives a stop and start of the event source no longer reschedules itself next to the polling chain of the new run - a periodic task cancelled while being re-armed is removed from the waiting set of the scheduled executor Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
…s an override changes their settings Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
…ecutors when started again while running - a per resource registration that races stop() is buffered until the next start instead of leaving a stale entry that blocks polling after a restart - the timer and per resource polling event sources resolve their executor again when started while running, which happens when the operator is started again after a failed start and replaces its executors Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
…s, and let in-flight work finish when stopping owned executors - PollingEventSource schedules its polling task again when started while running after the operator replaced its executors - TimerEventSource schedules the timers pending on the replaced executor again for their remaining delay - stopping a polling or timer event source that owns its executor no longer interrupts a poll or handler in flight Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Signed-off-by: Attila Mészáros <a_meszaros@apple.com>
aaa7b6c to
83cc809
Compare
There was a problem hiding this comment.
🟡 Changes recommended
Unresolved polling lifecycle races can overwrite fresh cache data or prevent polling after resource recreation, while the renamed public executor accessor breaks source and binary compatibility.
5 open issues
4 resolved since last review
Review effort: Balanced
…se races in the polling event sources - ExecutorServiceManager keeps the public scheduledExecutorService() of previous versions, deprecated, created on first use and replaced after a restart - PerResourcePollingEventSource drops the result of a fetch from a previous run instead of overwriting the cache of the current one - PollingEventSource ends start() cleanly when stopped during the initial fetch instead of failing on the shut down executor it owns Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
…ck in the fetching task Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
There was a problem hiding this comment.
🔵 Needs a closer look
shutdownNow() omits unstarted tasks from its return value, violating its contract, while a startup replay/deletion race can leave stale entries that prevent polling from restarting for recreated resources.
2 open issues
3 resolved since last review
Review effort: Balanced
…g the JVM alive Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
| } else if (!changes( | ||
| concurrentRetryAndRescheduleThreads, original.concurrentRetryAndRescheduleThreads())) { | ||
| return original.getRetryAndRescheduleExecutorService(); | ||
| } else { | ||
| return super.getRetryAndRescheduleExecutorService(); |
| if (future.isCancelled() || !future.isDone()) { | ||
| scheduleOnce(resourceID, Math.max(0, future.getDelay(TimeUnit.MILLISECONDS))); |


feat: virtual thread support
Adds a single flag that switches the framework's concurrent work over to virtual threads:
It can also be set with
josdk.use-virtual-threads. It is off by default, so nothing changes forexisting operators.
Why
Reconcilers spend almost all their time blocked on the Kubernetes API server or on external
systems. Virtual threads make that kind of blocking cheap, so concurrency limits can be raised far
above what is reasonable with platform threads, without sizing and tuning thread pools.
What changes when the flag is on
withConcurrentReconciliationThreads(n)still caps how many run at once (enforced with aSemaphore).withConcurrentWorkflowExecutorThreads(n).mainreturns after starting the operator and nothing else keeps the JVM running, callOperator.keepAlive()(already available, see the docs). The framework does not start such a thread on its own.Java 25 only
Virtual threads are supported on Java 25 and later (JEP 491
removed the carrier pinning inside
synchronizedblocks, which the codebase uses extensively; onJava 21 to 24 it can starve the carrier pool and deadlock startup). This is checked at runtime: on
an older JVM a warning is logged and platform threads are used, so one configuration works on every
Java version the operator runs on. The compilation target stays Java 17.
Scheduled executors (the larger part of this PR)
Previously
PollingEventSourceandTimerEventSourceeach created ajava.util.Timer, andPerResourcePollingEventSourcecreated its ownScheduledThreadPoolExecutor. Each cost a threadthat the operator neither sized nor shut down. They now use scheduled executors that are created
and shut down in a defined way:
PollingEventSourceandPerResourcePollingEventSourcedo not share an executor, so a slow poll of one source neverdelays another. Unless one is provided with
withExecutorServiceon the configuration builder,the source creates a single-thread daemon executor when it starts and shuts it down when it stops
(a new one on every start). With
useVirtualThreadson, that executor runs the polls on virtualthreads. A provided executor is used as is, is never shut down by the source, and may be shared
deliberately.
ConfigurationService.getRetryAndRescheduleExecutorService()backs the retry and rescheduletimer of every controller. It is managed by
ExecutorServiceManager, sized withconcurrentRetryAndRescheduleThreads()(default 2) orjosdk.retry-and-reschedule.concurrent-threads, and can be replaced throughConfigurationServiceOverrider. The operator shuts it down on stop and creates a new one whenstarted again.
reschedules don't delay termination), remove cancelled tasks eagerly, and let idle threads time
out after 60 seconds.
start()rather than at construction. The pollingand timer sources let a poll or handler already in flight finish when stopped, as the
Timerthey replace did, and the timer source re-arms its pending timers on the operator's new executor
when the operator is started again.
ExecutorServiceManager.scheduledExecutorService()stays available for compatibility,deprecated, as a general purpose executor created on first use. The framework itself no longer
uses it.
VirtualThreadScheduledExecutorThe JDK has no scheduled executor backed by virtual threads, so with the flag on the defaults are a
new package-private
VirtualThreadScheduledExecutor: one daemon platform thread does the timingand hands each fired task to a virtual-thread-per-task executor.
typically blocking calls.
withConcurrentRetryAndRescheduleThreadsonly applies to platformthreads.
on
shutdown(), and tasks that have already started are let finish.shutdownNow()interruptsrunning tasks.
ScheduledThreadPoolExecutordoes. Fixed-delay tasks re-arm from completion time.
Other changes
ExecutorUtilsInternal.ConfigLoaderandConfigurationServiceOverriderexpose the new settings(
josdk.use-virtual-threads,josdk.retry-and-reschedule.concurrent-threads).VirtualThreadScheduledExecutorhandles. They are ordered by remainingdelay but must stay distinct by identity, as the JDK's own scheduled futures.
Documentation
operations/configuration.md: new "Virtual Threads" section, which points to "Keeping the JVMAlive" for operators whose
mainreturns, and the new configuration keys.eventing.md: "Threading of the polling event sources", with the per-source executor and whoowns it.
Testing
Unit tests, with the virtual thread ones gated by
@EnabledForJreRange(min = JAVA_25):VirtualThreadScheduledExecutorTest: virtual thread execution, cancellation, fixed delay andfixed rate, overrun catch-up, shutdown semantics, unbounded concurrency.
VirtualThreadsTest,BoundedVirtualThreadExecutorAuditTest: bounded virtual thread executorand fallback on old JVMs.
VirtualThreadsStartupDeadlockTest: startup with virtual threads does not deadlock.ExecutorServiceManagerTest,ConfigurationServiceOverriderTest: lifecycle and overriding.PerResourcePollingEventSourceTest,PollingEventSourceTest,TimerEventSourceTest: each sourceowning its executor (and not sharing it), provided executors left running, virtual threads,
restart, stop and start ordering, polling resilience.
ExecutorUtilsInternalTest: idle pool threads time out and delayed tasks still run.Integration test:
VirtualThreadsITruns a full operator with the flag on.Known limitations and follow-ups