feat: schedule polling and retries on managed executors instead of Timers

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.

Author: csviri

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)

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 |

operator-framework-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.

operator-framework-core/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

operator-framework-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();