Only one scheduled event (re-schedul / retry) at one time

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/DefaultEventHandler.java

@@ -199,18 +199,15 @@ void eventProcessingFinished(
       // If a delete event present at this phase, it was received during reconciliation.
       // So we either removed the finalizer during reconciliation or we don't use finalizers.
       // Either way we don't want to retry.
-      if (retry != null && postExecutionControl.exceptionDuringExecution() &&
+      if (isRetryConfigured() && postExecutionControl.exceptionDuringExecution() &&
           !eventMarker.deleteEventPresent(customResourceID)) {
         handleRetryOnException(executionScope);
         // todo revisit monitoring since events are not present anymore
         // final var monitor = monitor(); executionScope.getEvents().forEach(e ->
         // monitor.failedEvent(executionScope.getCustomResourceID(), e));
         return;
       }
-
-      if (retry != null) {
-        handleSuccessfulExecutionRegardingRetry(executionScope);
-      }
+      cleanupOnSuccessfulExecution(executionScope);
       if (eventMarker.deleteEventPresent(customResourceID)) {
         cleanupForDeletedEvent(executionScope.getCustomResourceID());
       } else {
@@ -265,7 +262,7 @@ private boolean isCacheReadyForInstantReconciliation(ExecutionScope<R> execution
   private void reScheduleExecutionIfInstructed(PostExecutionControl<R> postExecutionControl,
       R customResource) {
     postExecutionControl.getReScheduleDelay().ifPresent(delay -> eventSourceManager
-        .getRetryTimerEventSource()
+        .getRetryAndRescheduleTimerEventSource()
         .scheduleOnce(customResource, delay));
   }
 
@@ -295,19 +292,21 @@ private void handleRetryOnException(ExecutionScope<R> executionScope) {
               delay,
               customResourceID);
           eventSourceManager
-              .getRetryTimerEventSource()
+              .getRetryAndRescheduleTimerEventSource()
               .scheduleOnce(executionScope.getCustomResource(), delay);
         },
         () -> log.error("Exhausted retries for {}", executionScope));
   }
 
-  private void handleSuccessfulExecutionRegardingRetry(ExecutionScope<R> executionScope) {
+  private void cleanupOnSuccessfulExecution(ExecutionScope<R> executionScope) {
     log.debug(
-        "Marking successful execution for resource: {}",
+        "Cleanup for successful execution for resource: {}",
         getName(executionScope.getCustomResource()));
-    retryState.remove(executionScope.getCustomResourceID());
+    if (isRetryConfigured()) {
+      retryState.remove(executionScope.getCustomResourceID());
+    }
     eventSourceManager
-        .getRetryTimerEventSource()
+        .getRetryAndRescheduleTimerEventSource()
         .cancelOnceSchedule(executionScope.getCustomResourceID());
   }
 
@@ -337,6 +336,10 @@ private void unsetUnderExecution(CustomResourceID customResourceUid) {
     underProcessing.remove(customResourceUid);
   }
 
+  private boolean isRetryConfigured() {
+    return retry != null;
+  }
+
   @Override
   public void close() {
     lock.lock();

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/DefaultEventSourceManager.java

@@ -22,7 +22,7 @@ public class DefaultEventSourceManager<R extends CustomResource<?, ?>>
   private final ReentrantLock lock = new ReentrantLock();
   private final Set<EventSource> eventSources = Collections.synchronizedSet(new HashSet<>());
   private DefaultEventHandler<R> defaultEventHandler;
-  private TimerEventSource<R> retryTimerEventSource;
+  private TimerEventSource<R> retryAndRescheduleTimerEventSource;
   private CustomResourceEventSource customResourceEventSource;
 
   DefaultEventSourceManager(DefaultEventHandler<R> defaultEventHandler) {
@@ -39,8 +39,8 @@ private void init(DefaultEventHandler<R> defaultEventHandler) {
     this.defaultEventHandler = defaultEventHandler;
     defaultEventHandler.setEventSourceManager(this);
 
-    this.retryTimerEventSource = new TimerEventSource<>();
-    registerEventSource(retryTimerEventSource);
+    this.retryAndRescheduleTimerEventSource = new TimerEventSource<>();
+    registerEventSource(retryAndRescheduleTimerEventSource);
   }
 
   @Override
@@ -98,8 +98,8 @@ public void cleanupForCustomResource(CustomResourceID customResourceUid) {
     }
   }
 
-  public TimerEventSource getRetryTimerEventSource() {
-    return retryTimerEventSource;
+  public TimerEventSource getRetryAndRescheduleTimerEventSource() {
+    return retryAndRescheduleTimerEventSource;
   }
 
   @Override

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/internal/CustomResourceEventSource.java

@@ -188,7 +188,7 @@ private T clone(T customResource) {
   /**
    * This will ensure that the next event received after this method is called will not be filtered
    * out.
-   * 
+   *
    * @param customResourceID - to which the event is related
    */
   public void whitelistNextEvent(CustomResourceID customResourceID) {

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/DefaultEventHandlerTest.java

@@ -53,7 +53,7 @@ class DefaultEventHandlerTest {
 
   @BeforeEach
   public void setup() {
-    when(defaultEventSourceManagerMock.getRetryTimerEventSource())
+    when(defaultEventSourceManagerMock.getRetryAndRescheduleTimerEventSource())
         .thenReturn(retryTimerEventSourceMock);
     defaultEventHandler.setEventSourceManager(defaultEventSourceManagerMock);
     defaultEventHandlerWithRetry.setEventSourceManager(defaultEventSourceManagerMock);
@@ -280,6 +280,17 @@ public void dontWhitelistsEventIfUpdatedEventInCache() {
     verify(mockCREventSource, times(0)).whitelistNextEvent(eq(crID));
   }
 
+  @Test
+  public void cancelScheduleOnceEventsOnSuccessfulExecution() {
+    var crID = new CustomResourceID("test-cr", TEST_NAMESPACE);
+    var cr = testCustomResource(crID);
+
+    defaultEventHandler.eventProcessingFinished(new ExecutionScope(cr, null),
+        PostExecutionControl.defaultDispatch());
+
+    verify(retryTimerEventSourceMock, times(1)).cancelOnceSchedule(eq(crID));
+  }
+
   private CustomResourceID eventAlreadyUnderProcessing() {
     when(eventDispatcherMock.handleExecution(any()))
         .then(