operator-framework · csviri · Sep 17, 2026
diff --git a/docs/content/en/docs/documentation/eventing.md b/docs/content/en/docs/documentation/eventing.md
@@ -222,6 +222,28 @@ is similar to `PerResourcePollingEventSource` except that, contrary to that even
 doesn't poll a specific API separately per resource, but periodically and independently of
 actually observed primary resources.
 
+#### Threading of the polling event sources
+
+Both polling event sources schedule their polls on an executor the operator shares between all of
+its polling event sources. A poll therefore only starts once one of that executor's threads is
+free, so if your operator registers many polling event sources, or if fetching your external
+resources is slow, size the pool accordingly with
+`ConfigurationServiceOverrider.withConcurrentScheduledTaskThreads` (4 threads by default):
+
+```java
+Operator operator = new Operator(overrider -> overrider.withConcurrentScheduledTaskThreads(20));
+```
+
+Retried and rescheduled reconciliations are triggered on a separate executor, so a slow poll can
+never delay them. It is sized with `withConcurrentRetryAndRescheduleThreads` (2 threads by
+default); few threads are needed there since triggering a reconciliation only enqueues an event
+for one of the reconciliation threads to pick up.
+
+Use `ConfigurationServiceOverrider.withScheduledExecutorService` to replace the polling executor
+altogether, or the `withExecutorService` method of the event source's own configuration builder to
+poll a single event source on an executor of its own. An executor provided that way is not managed
+by the operator: it is your responsibility to shut it down.
+
 #### Inbound event sources
 
 [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)

diff --git a/docs/content/en/docs/documentation/operations/configuration.md b/docs/content/en/docs/documentation/operations/configuration.md
@@ -281,6 +281,13 @@ All operator-level keys are prefixed with `josdk.`.
 |---|---|---|
 | `josdk.workflow.executor-threads` | `Integer` | Thread pool size for workflow execution |
 
+#### Scheduled Tasks
+
+| Key | Type | Description |
+|---|---|---|
+| `josdk.scheduled-tasks.concurrent-threads` | `Integer` | Thread pool size for the operator's scheduled tasks, i.e. the polling event sources |
+| `josdk.retry-and-reschedule.concurrent-threads` | `Integer` | Thread pool size for triggering retried and rescheduled reconciliations |
+
 #### Informer
 
 | Key | Type | Description |

diff --git a/...ework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java b/...ework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java
@@ -21,6 +21,7 @@
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.function.Consumer;
 
 import org.slf4j.Logger;
@@ -64,6 +65,20 @@ public interface ConfigurationService {
   /** The default number of threads used to process dependent workflows */
   int DEFAULT_WORKFLOW_EXECUTOR_THREAD_NUMBER = DEFAULT_RECONCILIATION_THREADS_NUMBER;
 
+  /**
+   * The default number of threads used to run the operator's scheduled tasks, i.e. the periodic
+   * polls of {@link io.javaoperatorsdk.operator.processing.event.source.polling.PollingEventSource}
+   * and {@link
+   * io.javaoperatorsdk.operator.processing.event.source.polling.PerResourcePollingEventSource}
+   */
+  int DEFAULT_SCHEDULED_TASK_THREADS_NUMBER = 4;
+
+  /**
+   * The default number of threads used to trigger the operator's retried and rescheduled
+   * reconciliations
+   */
+  int DEFAULT_RETRY_AND_RESCHEDULE_THREADS_NUMBER = 2;
+
   /**
    * Creates a new {@link ConfigurationService} instance used to configure an {@link
    * io.javaoperatorsdk.operator.Operator} instance, starting from the specified base configuration
@@ -219,6 +234,33 @@ default int concurrentWorkflowExecutorThreads() {
     return DEFAULT_WORKFLOW_EXECUTOR_THREAD_NUMBER;
   }
 
+  /**
+   * Number of threads the operator can spin out to run its scheduled (i.e. periodic or delayed)
+   * tasks with the default executor. These threads are shared by all the polling event sources of
+   * the operator, so this number should be raised when many, or slow, polling event sources are
+   * registered: a poll only starts once a thread is available, and a slow poll therefore delays the
+   * polls of the other event sources.
+   *
+   * @return the maximum number of concurrent scheduled task threads
+   * @since 5.6.0
+   */
+  default int concurrentScheduledTaskThreads() {
+    return DEFAULT_SCHEDULED_TASK_THREADS_NUMBER;
+  }
+
+  /**
+   * Number of threads the operator can spin out to trigger its retried and rescheduled
+   * reconciliations with the default executor. These threads are shared by all the controllers of
+   * the operator, but the tasks they run only enqueue an event for the reconciliation to happen on
+   * a reconciliation thread, so few of them are needed.
+   *
+   * @return the maximum number of concurrent retry and reschedule threads
+   * @since 5.6.0
+   */
+  default int concurrentRetryAndRescheduleThreads() {
+    return DEFAULT_RETRY_AND_RESCHEDULE_THREADS_NUMBER;
+  }
+
   /**
    * Override to provide a custom {@link Metrics} implementation
    *
@@ -249,6 +291,44 @@ default ExecutorService getWorkflowExecutorService() {
     return Executors.newFixedThreadPool(concurrentWorkflowExecutorThreads());
   }
 
+  /**
+   * Override to provide a custom {@link ScheduledExecutorService} implementation to change how the
+   * operator's scheduled (i.e. periodic or delayed) tasks are run. This executor is shared by all
+   * the polling event sources of the operator. The retried and rescheduled reconciliations run on
+   * an executor of their own, see {@link #getRetryAndRescheduleExecutorService()}, so that a slow
+   * poll can't delay them.
+   *
+   * <p>Note that the default implementation lets the executor discard the tasks that were scheduled
+   * for later when it is shut down, so that they don't delay the termination of the operator, and
+   * that it creates daemon threads so that a never stopped operator doesn't keep the JVM alive.
+   * Custom implementations are advised to do the same.
+   *
+   * @return the {@link ScheduledExecutorService} implementation to use to run scheduled tasks
+   * @since 5.6.0
+   */
+  default ScheduledExecutorService getScheduledExecutorService() {
+    return Utils.daemonScheduledThreadPool(
+        concurrentScheduledTaskThreads(), "josdk-scheduled-task");
+  }
+
+  /**
+   * Override to provide a custom {@link ScheduledExecutorService} implementation to change how the
+   * operator's retried and rescheduled reconciliations are triggered. This executor is kept
+   * separate from the one the polling event sources use, see {@link
+   * #getScheduledExecutorService()}, so that a slow poll can't delay a retry.
+   *
+   * <p>The same notes as for {@link #getScheduledExecutorService()} apply to custom
+   * implementations.
+   *
+   * @return the {@link ScheduledExecutorService} implementation to use to trigger retried and
+   *     rescheduled reconciliations
+   * @since 5.6.0
+   */
+  default ScheduledExecutorService getRetryAndRescheduleExecutorService() {
+    return Utils.daemonScheduledThreadPool(
+        concurrentRetryAndRescheduleThreads(), "josdk-retry-reschedule");
+  }
+
   /**
    * Determines whether the associated Kubernetes client should be closed when the associated {@link
    * io.javaoperatorsdk.operator.Operator} is stopped.

diff --git a/...e/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java b/...e/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java
@@ -21,6 +21,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.function.Function;
 
 import org.slf4j.Logger;
@@ -45,11 +46,15 @@ public class ConfigurationServiceOverrider {
   private Boolean checkCR;
   private Integer concurrentReconciliationThreads;
   private Integer concurrentWorkflowExecutorThreads;
+  private Integer concurrentScheduledTaskThreads;
+  private Integer concurrentRetryAndRescheduleThreads;
   private Cloner cloner;
   private Boolean closeClientOnStop;
   private KubernetesClient client;
   private ExecutorService executorService;
   private ExecutorService workflowExecutorService;
+  private ScheduledExecutorService scheduledExecutorService;
+  private ScheduledExecutorService retryAndRescheduleExecutorService;
   private LeaderElectionConfiguration leaderElectionConfiguration;
   private String clusterScopedEventNamespace;
   private EventRecorder eventRecorder;
@@ -86,6 +91,34 @@ public ConfigurationServiceOverrider withConcurrentWorkflowExecutorThreads(int t
     return this;
   }
 
+  /**
+   * Sets the number of threads used to run the operator's scheduled (i.e. periodic or delayed)
+   * tasks, which are shared by all its polling event sources.
+   *
+   * @param threadNumber the maximum number of concurrent scheduled task threads
+   * @return this {@link ConfigurationServiceOverrider} for chained customization
+   * @see ConfigurationService#concurrentScheduledTaskThreads()
+   * @since 5.6.0
+   */
+  public ConfigurationServiceOverrider withConcurrentScheduledTaskThreads(int threadNumber) {
+    this.concurrentScheduledTaskThreads = threadNumber;
+    return this;
+  }
+
+  /**
+   * Sets the number of threads used to trigger the operator's retried and rescheduled
+   * reconciliations, which are shared by all its controllers.
+   *
+   * @param threadNumber the maximum number of concurrent retry and reschedule threads
+   * @return this {@link ConfigurationServiceOverrider} for chained customization
+   * @see ConfigurationService#concurrentRetryAndRescheduleThreads()
+   * @since 5.6.0
+   */
+  public ConfigurationServiceOverrider withConcurrentRetryAndRescheduleThreads(int threadNumber) {
+    this.concurrentRetryAndRescheduleThreads = threadNumber;
+    return this;
+  }
+
   @SuppressWarnings("rawtypes")
   public ConfigurationServiceOverrider withDependentResourceFactory(
       DependentResourceFactory dependentResourceFactory) {
@@ -119,6 +152,37 @@ public ConfigurationServiceOverrider withWorkflowExecutorService(
     return this;
   }
 
+  /**
+   * Replaces the executor used to run the operator's scheduled (i.e. periodic or delayed) tasks,
+   * which are shared by all its polling event sources.
+   *
+   * @param scheduledExecutorService the executor to run scheduled tasks on
+   * @return this {@link ConfigurationServiceOverrider} for chained customization
+   * @see ConfigurationService#getScheduledExecutorService()
+   * @since 5.6.0
+   */
+  public ConfigurationServiceOverrider withScheduledExecutorService(
+      ScheduledExecutorService scheduledExecutorService) {
+    this.scheduledExecutorService = scheduledExecutorService;
+    return this;
+  }
+
+  /**
+   * Replaces the executor used to trigger the operator's retried and rescheduled reconciliations,
+   * which is shared by all its controllers.
+   *
+   * @param retryAndRescheduleExecutorService the executor to trigger retried and rescheduled
+   *     reconciliations on
+   * @return this {@link ConfigurationServiceOverrider} for chained customization
+   * @see ConfigurationService#getRetryAndRescheduleExecutorService()
+   * @since 5.6.0
+   */
+  public ConfigurationServiceOverrider withRetryAndRescheduleExecutorService(
+      ScheduledExecutorService retryAndRescheduleExecutorService) {
+    this.retryAndRescheduleExecutorService = retryAndRescheduleExecutorService;
+    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.
@@ -312,6 +376,28 @@ public int concurrentWorkflowExecutorThreads() {
             original.concurrentWorkflowExecutorThreads());
       }
 
+      @Override
+      public int concurrentScheduledTaskThreads() {
+        return Utils.ensureValid(
+            overriddenValueOrDefault(
+                concurrentScheduledTaskThreads,
+                ConfigurationService::concurrentScheduledTaskThreads),
+            "maximum scheduled task threads",
+            1,
+            original.concurrentScheduledTaskThreads());
+      }
+
+      @Override
+      public int concurrentRetryAndRescheduleThreads() {
+        return Utils.ensureValid(
+            overriddenValueOrDefault(
+                concurrentRetryAndRescheduleThreads,
+                ConfigurationService::concurrentRetryAndRescheduleThreads),
+            "maximum retry and reschedule threads",
+            1,
+            original.concurrentRetryAndRescheduleThreads());
+      }
+
       @Override
       public Metrics getMetrics() {
         return overriddenValueOrDefault(metrics, ConfigurationService::getMetrics);
@@ -340,6 +426,24 @@ public ExecutorService getWorkflowExecutorService() {
         }
       }
 
+      @Override
+      public ScheduledExecutorService getScheduledExecutorService() {
+        if (scheduledExecutorService != null) {
+          return scheduledExecutorService;
+        } else {
+          return super.getScheduledExecutorService();
+        }
+      }
+
+      @Override
+      public ScheduledExecutorService getRetryAndRescheduleExecutorService() {
+        if (retryAndRescheduleExecutorService != null) {
+          return retryAndRescheduleExecutorService;
+        } else {
+          return super.getRetryAndRescheduleExecutorService();
+        }
+      }
+
       @Override
       public Optional<LeaderElectionConfiguration> getLeaderElectionConfiguration() {
         return leaderElectionConfiguration != null

diff --git a/...ork-core/src/main/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManager.java b/...ork-core/src/main/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManager.java
@@ -42,6 +42,7 @@ public class ExecutorServiceManager {
   private ExecutorService workflowExecutor;
   private ExecutorService cachingExecutorService;
   private ScheduledExecutorService scheduledExecutorService;
+  private ScheduledExecutorService retryAndRescheduleExecutorService;
   private boolean started;
   private ConfigurationService configurationService;
 
@@ -128,30 +129,54 @@ public ExecutorService cachingExecutorService() {
     return cachingExecutorService;
   }
 
+  /**
+   * The executor the operator runs its scheduled (i.e. periodic or delayed) tasks on, shared by its
+   * polling event sources. Note that it is only valid while the manager is started: it is shut down
+   * by {@link #stop(Duration)} and replaced by a fresh one on the next {@link
+   * #start(ConfigurationService)}, so callers should retrieve it when they start rather than hold
+   * on to it.
+   *
+   * @return the executor to run scheduled tasks on
+   */
   public ScheduledExecutorService scheduledExecutorService() {
     return scheduledExecutorService;
   }
 
+  /**
+   * The executor the operator triggers its retried and rescheduled reconciliations on, kept
+   * separate from {@link #scheduledExecutorService()} so that a slow poll can't delay a retry. The
+   * same lifecycle caveat as for {@link #scheduledExecutorService()} applies.
+   *
+   * @return the executor to trigger retried and rescheduled reconciliations on
+   */
+  public ScheduledExecutorService retryAndRescheduleExecutorService() {
+    return retryAndRescheduleExecutorService;
+  }
+
   public synchronized void start(ConfigurationService configurationService) {
     if (!started) {
       this.configurationService = configurationService; // used to lazy init workflow executor
       this.cachingExecutorService = Executors.newCachedThreadPool();
-      this.scheduledExecutorService = Executors.newScheduledThreadPool(0);
+      this.scheduledExecutorService = configurationService.getScheduledExecutorService();
+      this.retryAndRescheduleExecutorService =
+          configurationService.getRetryAndRescheduleExecutorService();
       this.executor = new InstrumentedExecutorService(configurationService.getExecutorService());
       started = true;
     }
   }
 
   public synchronized void stop(Duration gracefulShutdownTimeout) {
-    var parallelExec = Executors.newFixedThreadPool(4);
+    var shutdowns =
+        List.of(
+            shutdown(executor, gracefulShutdownTimeout),
+            shutdown(workflowExecutor, gracefulShutdownTimeout),
+            shutdown(cachingExecutorService, gracefulShutdownTimeout),
+            shutdown(scheduledExecutorService, gracefulShutdownTimeout),
+            shutdown(retryAndRescheduleExecutorService, gracefulShutdownTimeout));
+    var parallelExec = Executors.newFixedThreadPool(shutdowns.size());
     try {
       log.debug("Closing executor");
-      parallelExec.invokeAll(
-          List.of(
-              shutdown(executor, gracefulShutdownTimeout),
-              shutdown(workflowExecutor, gracefulShutdownTimeout),
-              shutdown(cachingExecutorService, gracefulShutdownTimeout),
-              shutdown(scheduledExecutorService, gracefulShutdownTimeout)));
+      parallelExec.invokeAll(shutdowns);
     } catch (InterruptedException e) {
       log.debug("Exception closing executor: {}", e.getLocalizedMessage());
       Thread.currentThread().interrupt();