Repository navigation
Conversation
…mers PollingEventSource and TimerEventSource each created a java.util.Timer, and PerResourcePollingEventSource a ScheduledThreadPoolExecutor of its own, so every polling event source and every controller cost a thread that the operator neither sized nor shut down. They now schedule on executors managed by ExecutorServiceManager, which builds them from ConfigurationService: - getScheduledExecutorService() backs the polling event sources - getRetryAndRescheduleExecutorService() backs the retry and reschedule timer of every controller, kept separate so that a slow poll cannot delay a retry Both are sized by concurrentScheduledTaskThreads() (4 by default) and concurrentRetryAndRescheduleThreads() (2 by default), overridable programmatically or through the josdk.scheduled-tasks.concurrent-threads and josdk.retry-and-reschedule.concurrent-threads keys. Both default to daemon threads, discard the tasks scheduled for later on shutdown so they don't hold up the termination of the operator, and drop cancelled tasks eagerly. The previous Executors.newScheduledThreadPool(0) was effectively single threaded and created non daemon threads. An event source resolves its executor on every start rather than at creation time, since the manager replaces its pools when the operator is restarted, and only shuts one down if it created it itself. That also fixes PerResourcePollingEventSource shutting down a user supplied executor and not registering its tasks again after a restart. PollingEventSource gains a constructor taking an EventSourceContext, the one without it is deprecated, and both polling configurations accept an executor to run a single event source on a pool of its own.
…rator replaces its executors
… resource polling, clarify executor ownership in docs
…ed after stop during start
Signed-off-by: Attila Mészáros <a_meszaros@apple.com>
Signed-off-by: Attila Mészáros <a_meszaros@apple.com>
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.
…ording and desired state aspects - bounded virtual thread executor rejects non-positive concurrency and returns tasks that never started from shutdownNow - event sink aggregation also updates type, reason and action - long supplied event names are shortened with a hash so they stay distinct - events for cluster scoped resources in an overridden namespace carry an eventTime - desired state aspects can look up the actual resource of their dependent, receive the user's bulk dependent, and tolerate a missing controller configuration
|
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
 |
This was referenced Oct 27, 2026
Contributor
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
- remove leftover platform thread executor assignments that overrode the virtual thread based caching executor in ExecutorServiceManager - buffer per resource polling registrations that arrive before the event source started and replay them once the executor is resolved - close the race between execute and shutdownNow of the bounded virtual thread executor with a shared lifecycle lock - reject aggregating events that carry an eventTime when type, reason or action differ - keep the JVM alive with a non-daemon thread while virtual threads are in use - run the startup deadlock test on Java 25 or later only
The event recorder aggregation, long name and cluster scoped event namespace fixes now live in fix/event-recorder-correctness.
…nternal Add a virtual thread backed scheduled executor and related configuration, docs and tests. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
…resilience - use the daemon scheduled thread pool for standalone timer and polling executors - resolve the timer executor before marking the event source as running - keep polling a resource after its fetcher failed - exclude the intentional compareTo/equals mismatch of scheduled handles from SpotBugs Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Contributor
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
4 resolved since last review
Review effort: Balanced
Comment on lines
+124
to
+128
| cancelWaiting(true); | ||
| timer.shutdownNow(); | ||
| for (var task : worker.shutdownNow()) { | ||
| if (task instanceof Dispatched dispatched) { | ||
| dispatched.handle.cancel(false); |
Comment on lines
+200
to
+202
| if (result.isDone()) { | ||
| timerFuture.cancel(false); | ||
| } |
Comment on lines
+218
to
+222
| if (result.isDone()) { | ||
| return; | ||
| } | ||
| runner = Thread.currentThread(); | ||
| try { |
| } | ||
| final long delay; | ||
| if (period > 0) { | ||
| delay = Math.max(0, triggerTime + period - System.nanoTime()); |
Comment on lines
+112
to
+114
| var resource = registrationsBeforeStart.remove(primaryID); | ||
| if (resource != null) { | ||
| checkAndRegisterTask(resource); |
Comment on lines
+156
to
+157
| if (!isRunning() && scheduledFutures.remove(primaryID, scheduledFuture)) { | ||
| scheduledFuture.cancel(true); |
|
|
||
| eventSource.start(); | ||
| eventSource.scheduleOnce(ResourceID.fromResource(TestUtils.testCustomResource()), PERIOD); | ||
| untilAsserted(() -> assertThat(handler.events).hasSize(1)); |
Contributor
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… · Newstop()and a subsequentstart(), this check sees the source… · NewshutdownNow()omits scheduled tasks that never started: it discards both waiting timers and… · NewisDone()check but beforerunneris assigned, so… · Newrearm()clamps the delay to zero andarm()resets… · NewCapturingEventHandler… · New
Review effort: Balanced
Comment on lines
+219
to
+220
| if (VirtualThreads.shouldUse(configurationService.useVirtualThreads())) { | ||
| startKeepAliveThread(); |
Comment on lines
+112
to
+114
| var resource = registrationsBeforeStart.remove(primaryID); | ||
| if (resource != null) { | ||
| checkAndRegisterTask(resource); |
Comment on lines
+156
to
+157
| if (!isRunning() && scheduledFutures.remove(primaryID, scheduledFuture)) { | ||
| scheduledFuture.cancel(true); |
Comment on lines
+124
to
+128
| cancelWaiting(true); | ||
| timer.shutdownNow(); | ||
| for (var task : worker.shutdownNow()) { | ||
| if (task instanceof Dispatched dispatched) { | ||
| dispatched.handle.cancel(false); |
Comment on lines
+217
to
+223
| private void runOnWorker() { | ||
| if (result.isDone()) { | ||
| return; | ||
| } | ||
| runner = Thread.currentThread(); | ||
| try { | ||
| if (period == 0) { |
| } | ||
| final long delay; | ||
| if (period > 0) { | ||
| delay = Math.max(0, triggerTime + period - System.nanoTime()); |
|
|
||
| eventSource.start(); | ||
| eventSource.scheduleOnce(ResourceID.fromResource(TestUtils.testCustomResource()), PERIOD); | ||
| untilAsserted(() -> assertThat(handler.events).hasSize(1)); |
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Introduces feature flag to use virtual threads instead of standard
ThreadPools.Thus, basically only replaces internal the thread pools, with ones that use virtual threads.
The concurrency limit also applies for this new mode (using
Semaphore).Note that for now we aim for Java 25, and the intention is that only that will be officially supported, since
synchronizedkeyword is used extensively in the codebase.TODO: