From 4e1cf6309f06b0b28062660dc56dc2ce4db2f2e5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Attila=20M=C3=A9sz=C3=A1ros?= Date: Sun, 8 Nov 2026 11:36:02 -0800 Subject: [PATCH 1/2] fix: address code review findings on virtual thread support - deprecate TimerEventSource(String, boolean), as documented in the migration guide - correct @since of the desired state aspect APIs to 5.7.0 - extract the executor creation shared by the polling event sources - remove comments restating the code --- .../api/config/ConfigurationService.java | 2 +- .../config/ConfigurationServiceOverrider.java | 4 +- .../BulkDependentResourceReconciler.java | 3 -- .../informer/pool/AbstractInformerPool.java | 1 - .../PerResourcePollingEventSource.java | 21 +-------- .../source/polling/PollingEventSource.java | 28 +++--------- .../source/polling/PollingExecutors.java | 43 +++++++++++++++++++ .../event/source/timer/TimerEventSource.java | 7 ++- 8 files changed, 58 insertions(+), 51 deletions(-) create mode 100644 operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingExecutors.java 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 6a7f2f93d5..b467a9c88b 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 @@ -653,7 +653,7 @@ default InformerPool informerPool() { * the operator creates or updates. * * @return the list of aspects to apply to computed desired states, empty by default - * @since 5.6.0 + * @since 5.7.0 */ default List desiredStateAspects() { return List.of(); 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 c95b9cc54a..2330a83ee4 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 @@ -297,7 +297,7 @@ public ConfigurationServiceOverrider withInformerPool(InformerPool informerPool) * * @param desiredStateAspects the aspects to apply, in the order in which they should be applied * @return this {@link ConfigurationServiceOverrider} for chained customization - * @since 5.6.0 + * @since 5.7.0 */ public ConfigurationServiceOverrider withDesiredStateAspects( List desiredStateAspects) { @@ -312,7 +312,7 @@ public ConfigurationServiceOverrider withDesiredStateAspects( * * @param desiredStateAspects the aspects to append, in the order in which they should be applied * @return this {@link ConfigurationServiceOverrider} for chained customization - * @since 5.6.0 + * @since 5.7.0 */ public ConfigurationServiceOverrider addDesiredStateAspects( DesiredStateAspect... desiredStateAspects) { diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/BulkDependentResourceReconciler.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/BulkDependentResourceReconciler.java index 2428eeba96..53ccf67cb2 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/BulkDependentResourceReconciler.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/BulkDependentResourceReconciler.java @@ -117,9 +117,6 @@ public R update(R actual, R desired, P primary, Context

context) { @Override public Result match(R resource, P primary, Context

context) { - // retrieve the desired state via the context so that it is processed the same way as for - // non-bulk dependents, in particular so that configured DesiredStateAspects are applied - // before matching return bulkDependentResource.match(resource, getOrComputeDesired(context), primary, context); } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AbstractInformerPool.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AbstractInformerPool.java index 21464b378b..ab9859518e 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AbstractInformerPool.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AbstractInformerPool.java @@ -117,7 +117,6 @@ protected SharedIndexInformer createInformer(InformerClassifier classifier) { Optional.ofNullable(classifier.itemStore()).ifPresent(informer::itemStore); if (classifier.withoutNamespaceIndex()) { - // the framework only reads the indexes registered through the resource cache, never this one informer.removeNamespaceIndex(); } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSource.java index 10c8cba913..6a4fe5ca79 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSource.java @@ -31,8 +31,6 @@ import io.fabric8.kubernetes.api.model.HasMetadata; import io.javaoperatorsdk.operator.OperatorException; -import io.javaoperatorsdk.operator.api.config.ConfigurationService; -import io.javaoperatorsdk.operator.api.config.ExecutorServiceManager; import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext; import io.javaoperatorsdk.operator.processing.event.ResourceID; import io.javaoperatorsdk.operator.processing.event.source.Cache; @@ -80,25 +78,11 @@ public PerResourcePollingEventSource( this.resourceFetcher = config.resourceFetcher(); this.registerPredicate = config.registerPredicate(); this.configuredExecutorService = config.executorService(); - var configurationService = configurationServiceOf(context); - this.useVirtualThreads = - configurationService != null && configurationService.useVirtualThreads(); + this.useVirtualThreads = PollingExecutors.useVirtualThreads(context); this.ownsExecutorService = configuredExecutorService == null; this.period = config.defaultPollingPeriod(); } - /** - * The configuration of the operator the event source belongs to, or {@code null} if it doesn't - * belong to one, which only happens when the event source is used standalone, outside an - * operator. - */ - private static ConfigurationService configurationServiceOf(EventSourceContext context) { - final var controllerConfiguration = context.getControllerConfiguration(); - return controllerConfiguration == null - ? null - : controllerConfiguration.getConfigurationService(); - } - @Override public void start() throws OperatorException { if (isRunning()) { @@ -119,8 +103,7 @@ private ScheduledExecutorService resolveExecutorService() { if (configuredExecutorService != null) { return configuredExecutorService; } - return ExecutorServiceManager.newScheduledExecutorService( - 1, useVirtualThreads, "josdk-polling-" + name()); + return PollingExecutors.newExecutor(name(), useVirtualThreads); } private Set getAndCacheResource(P primary, boolean fromGetter) { diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSource.java index a33e6cb44e..b2272822e7 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSource.java @@ -30,7 +30,6 @@ import io.fabric8.kubernetes.api.model.HasMetadata; import io.javaoperatorsdk.operator.OperatorException; import io.javaoperatorsdk.operator.api.config.ConfigurationService; -import io.javaoperatorsdk.operator.api.config.ExecutorServiceManager; import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext; import io.javaoperatorsdk.operator.health.Status; import io.javaoperatorsdk.operator.processing.event.ResourceID; @@ -89,7 +88,7 @@ public class PollingEventSource */ public PollingEventSource( Class resourceClass, EventSourceContext

context, PollingConfiguration config) { - this(resourceClass, config, configurationServiceOf(context)); + this(resourceClass, config, PollingExecutors.useVirtualThreads(context)); } /** @@ -100,34 +99,19 @@ public PollingEventSource( */ @Deprecated(since = "5.7.0") public PollingEventSource(Class resourceClass, PollingConfiguration config) { - this(resourceClass, config, null); + this(resourceClass, config, false); } private PollingEventSource( - Class resourceClass, - PollingConfiguration config, - ConfigurationService configurationService) { + Class resourceClass, PollingConfiguration config, boolean useVirtualThreads) { super(config.name(), resourceClass, config.resourceIDMapper()); this.genericResourceFetcher = config.genericResourceFetcher(); this.period = config.period(); this.configuredExecutorService = config.executorService(); - this.useVirtualThreads = - configurationService != null && configurationService.useVirtualThreads(); + this.useVirtualThreads = useVirtualThreads; this.ownsExecutorService = configuredExecutorService == null; } - /** - * The configuration of the operator the event source belongs to, or {@code null} if it doesn't - * belong to one, which only happens when the event source is used standalone, outside an - * operator. - */ - private static ConfigurationService configurationServiceOf(EventSourceContext context) { - final var controllerConfiguration = context.getControllerConfiguration(); - return controllerConfiguration == null - ? null - : controllerConfiguration.getConfigurationService(); - } - @Override public void start() throws OperatorException { if (pollingTask != null) { @@ -187,8 +171,7 @@ private ScheduledExecutorService resolveExecutorService() { if (configuredExecutorService != null) { return configuredExecutorService; } - return ExecutorServiceManager.newScheduledExecutorService( - 1, useVirtualThreads, "josdk-polling-" + name()); + return PollingExecutors.newExecutor(name(), useVirtualThreads); } protected synchronized void getStateAndFillCache() { @@ -204,7 +187,6 @@ public interface GenericResourceFetcher { public void stop() throws OperatorException { super.stop(); if (pollingTask != null) { - // as with the java.util.Timer this replaces, an ongoing poll is left to finish pollingTask.cancel(false); pollingTask = null; } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingExecutors.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingExecutors.java new file mode 100644 index 0000000000..02dc8b245e --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingExecutors.java @@ -0,0 +1,43 @@ +/* + * 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.processing.event.source.polling; + +import java.util.concurrent.ScheduledExecutorService; + +import io.javaoperatorsdk.operator.api.config.ExecutorServiceManager; +import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext; + +final class PollingExecutors { + + private PollingExecutors() {} + + /** + * Whether the operator the event source belongs to is configured to use virtual threads. The + * controller configuration is {@code null} when the event source is used standalone, outside an + * operator. + */ + static boolean useVirtualThreads(EventSourceContext context) { + final var controllerConfiguration = context.getControllerConfiguration(); + return controllerConfiguration != null + && controllerConfiguration.getConfigurationService().useVirtualThreads(); + } + + static ScheduledExecutorService newExecutor(String eventSourceName, boolean useVirtualThreads) { + return ExecutorServiceManager.newScheduledExecutorService( + 1, useVirtualThreads, "josdk-polling-" + eventSourceName); + } +} diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/timer/TimerEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/timer/TimerEventSource.java index c244242a62..2aecabaa6b 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/timer/TimerEventSource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/timer/TimerEventSource.java @@ -54,6 +54,11 @@ public TimerEventSource() { this.ownsExecutorService = true; } + /** + * @deprecated use {@link #TimerEventSource(String, boolean, Supplier)} instead, passing {@code + * null} as the supplier to keep the previous behavior. + */ + @Deprecated(since = "5.7.0") public TimerEventSource(String name, boolean triggerReconcilerOnAllEvents) { this(name, triggerReconcilerOnAllEvents, null); } @@ -134,8 +139,6 @@ public void onResourceDeleted(R resource) { public void cancelOnceSchedule(ResourceID customResourceUid) { var scheduled = onceTasks.remove(customResourceUid); if (scheduled != null) { - // as with the java.util.TimerTask this replaces, a task that is already running is left to - // finish scheduled.cancel(false); } } From 9758422cfbad328c633c77f9e7ce4f1e36405461 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Attila=20M=C3=A9sz=C3=A1ros?= Date: Mon, 9 Nov 2026 13:28:10 -0700 Subject: [PATCH 2/2] fix: keep null check on configuration service when resolving virtual thread setting --- .../processing/event/source/polling/PollingExecutors.java | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingExecutors.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingExecutors.java index 02dc8b245e..7e949881e4 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingExecutors.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingExecutors.java @@ -32,8 +32,11 @@ private PollingExecutors() {} */ static boolean useVirtualThreads(EventSourceContext context) { final var controllerConfiguration = context.getControllerConfiguration(); - return controllerConfiguration != null - && controllerConfiguration.getConfigurationService().useVirtualThreads(); + if (controllerConfiguration == null) { + return false; + } + final var configurationService = controllerConfiguration.getConfigurationService(); + return configurationService != null && configurationService.useVirtualThreads(); } static ScheduledExecutorService newExecutor(String eventSourceName, boolean useVirtualThreads) {