fix: keep DEFAULT_EXECUTOR_THREAD_NUMBER and withdraw polls scheduled after stop

Author: csviri

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingConfiguration.java

@@ -35,6 +35,13 @@ public record PerResourcePollingConfiguration<R, P extends HasMetadata, ID>(
     Predicate<P> registerPredicate,
     Duration defaultPollingPeriod) {
 
+  /**
+   * @deprecated not used anymore, polls now run on the executor the operator shares between all its
+   *     scheduled tasks unless an executor is explicitly configured
+   */
+  @Deprecated(forRemoval = true)
+  public static final int DEFAULT_EXECUTOR_THREAD_NUMBER = 1;
+
   public PerResourcePollingConfiguration(
       String name,
       ScheduledExecutorService executorService,

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSource.java

@@ -137,6 +137,12 @@ private void scheduleNextExecution(P primary, Set<R> actualResources) {
             executorService.schedule(
                 new FetchingExecutor(primaryID), fetchDuration.toMillis(), TimeUnit.MILLISECONDS);
     scheduledFutures.put(primaryID, scheduledFuture);
+    // a fetch that was already running when the event source got stopped can get here after stop
+    // cleared the map; since the shared executor stays alive, the task has to be withdrawn,
+    // otherwise the stale entry would prevent the resource from being registered on a restart
+    if (!isRunning() && scheduledFutures.remove(primaryID, scheduledFuture)) {
+      scheduledFuture.cancel(true);
+    }
   }
 
   @Override