retry tests, fixes

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

@@ -108,12 +108,13 @@ void eventProcessingFinished(
       unsetUnderExecution(executionScope.getCustomResourceUid());
 
       if (retry != null && postExecutionControl.exceptionDuringExecution()) {
-        handleRetryOnException(executionScope, postExecutionControl);
+        handleRetryOnException(executionScope);
         return;
-      } else if (retry != null) {
-        handleSuccessfulExecutionRegardingRetry(executionScope);
       }
 
+      if (retry != null) {
+        markSuccessfulExecutionRegardingRetry(executionScope);
+      }
       if (containsCustomResourceDeletedEvent(executionScope.getEvents())) {
         cleanupAfterDeletedEvent(executionScope.getCustomResourceUid());
       } else {
@@ -130,25 +131,24 @@ void eventProcessingFinished(
    * events (received meanwhile retry is in place or already in buffer) instantly or always wait
    * according to the retry timing if there was an exception.
    */
-  private void handleRetryOnException(
-      ExecutionScope executionScope, PostExecutionControl postExecutionControl) {
+  private void handleRetryOnException(ExecutionScope executionScope) {
     RetryExecution execution = getOrInitRetryExecution(executionScope);
     boolean newEventsExists = eventBuffer.newEventsExists(executionScope.getCustomResourceUid());
     eventBuffer.putBackEvents(executionScope.getCustomResourceUid(), executionScope.getEvents());
 
-    Optional<Long> nextDelay = execution.nextDelay();
     if (newEventsExists) {
       executeBufferedEvents(executionScope.getCustomResourceUid());
       return;
     }
+    Optional<Long> nextDelay = execution.nextDelay();
     nextDelay.ifPresent(
         delay ->
             defaultEventSourceManager
                 .getRetryTimerEventSource()
                 .scheduleOnce(executionScope.getCustomResource(), delay));
   }
 
-  private void handleSuccessfulExecutionRegardingRetry(ExecutionScope executionScope) {
+  private void markSuccessfulExecutionRegardingRetry(ExecutionScope executionScope) {
     retryState.remove(executionScope.getCustomResourceUid());
     defaultEventSourceManager
         .getRetryTimerEventSource()

operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/EventBuffer.java

@@ -14,11 +14,13 @@ public void addEvent(Event event) {
   }
 
   public boolean newEventsExists(String resourceId) {
-    return !events.get(resourceId).isEmpty();
+    return events.get(resourceId) != null && !events.get(resourceId).isEmpty();
   }
 
   public void putBackEvents(String resourceUid, List<Event> oldEvents) {
-    events.get(resourceUid).addAll(0, oldEvents);
+    List<Event> crEvents =
+        events.computeIfAbsent(resourceUid, (id) -> new ArrayList<>(oldEvents.size()));
+    crEvents.addAll(0, oldEvents);
   }
 
   public boolean containsEvents(String customResourceId) {

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

@@ -9,7 +9,10 @@
 import io.javaoperatorsdk.operator.processing.event.Event;
 import io.javaoperatorsdk.operator.processing.event.internal.CustomResourceEvent;
 import io.javaoperatorsdk.operator.processing.event.internal.TimerEvent;
+import io.javaoperatorsdk.operator.processing.event.internal.TimerEventSource;
+import io.javaoperatorsdk.operator.processing.retry.GenericRetry;
 import io.javaoperatorsdk.operator.sample.simple.TestCustomResource;
+import java.util.Arrays;
 import java.util.List;
 import java.util.UUID;
 import org.junit.jupiter.api.BeforeEach;
@@ -23,14 +26,26 @@ class DefaultEventHandlerTest {
   public static final int SEPARATE_EXECUTION_TIMEOUT = 450;
   private EventDispatcher eventDispatcherMock = mock(EventDispatcher.class);
   private CustomResourceCache customResourceCache = new CustomResourceCache();
-  private DefaultEventHandler defaultEventHandler =
-      new DefaultEventHandler(customResourceCache, eventDispatcherMock, "Test", null);
   private DefaultEventSourceManager defaultEventSourceManagerMock =
       mock(DefaultEventSourceManager.class);
+  private TimerEventSource retryTimerEventSourceMock = mock(TimerEventSource.class);
+
+  private DefaultEventHandler defaultEventHandler =
+      new DefaultEventHandler(customResourceCache, eventDispatcherMock, "Test", null);
+
+  private DefaultEventHandler defaultEventHandlerWithRetry =
+      new DefaultEventHandler(
+          customResourceCache,
+          eventDispatcherMock,
+          "Test",
+          GenericRetry.defaultLimitedExponentialRetry());
 
   @BeforeEach
   public void setup() {
+    when(defaultEventSourceManagerMock.getRetryTimerEventSource())
+        .thenReturn(retryTimerEventSourceMock);
     defaultEventHandler.setDefaultEventSourceManager(defaultEventSourceManagerMock);
+    defaultEventHandlerWithRetry.setDefaultEventSourceManager(defaultEventSourceManagerMock);
   }
 
   @Test
@@ -85,13 +100,57 @@ public void cleanUpAfterDeleteEvent() {
     String uid = customResource.getMetadata().getUid();
 
     defaultEventHandler.handleEvent(event);
-    // todo awaitility?
+
     waitMinimalTime();
 
     verify(defaultEventSourceManagerMock, times(1)).cleanup(uid);
     assertThat(customResourceCache.getLatestResource(uid)).isNotPresent();
   }
 
+  @Test
+  public void schedulesAnEventRetryOnException() {
+    Event event = prepareCREvent();
+    TestCustomResource customResource = testCustomResource();
+
+    ExecutionScope executionScope = new ExecutionScope(Arrays.asList(event), customResource);
+    PostExecutionControl postExecutionControl =
+        PostExecutionControl.exceptionDuringExecution(new RuntimeException("test"));
+
+    defaultEventHandlerWithRetry.eventProcessingFinished(executionScope, postExecutionControl);
+
+    verify(retryTimerEventSourceMock, times(1))
+        .scheduleOnce(eq(customResource), eq(GenericRetry.DEFAULT_INITIAL_INTERVAL));
+  }
+
+  @Test
+  public void executesTheControllerInstantlyAfterErrorIfEventsBuffered() {
+    Event event = prepareCREvent();
+    TestCustomResource customResource = testCustomResource();
+    customResource.getMetadata().setUid(event.getRelatedCustomResourceUid());
+    ExecutionScope executionScope = new ExecutionScope(Arrays.asList(event), customResource);
+    PostExecutionControl postExecutionControl =
+        PostExecutionControl.exceptionDuringExecution(new RuntimeException("test"));
+
+    // start processing an event
+    defaultEventHandlerWithRetry.handleEvent(event);
+    // buffer an another event
+    defaultEventHandlerWithRetry.handleEvent(event);
+    verify(eventDispatcherMock, timeout(SEPARATE_EXECUTION_TIMEOUT).times(1))
+        .handleExecution(any());
+
+    defaultEventHandlerWithRetry.eventProcessingFinished(executionScope, postExecutionControl);
+
+    ArgumentCaptor<ExecutionScope> executionScopeArgumentCaptor =
+        ArgumentCaptor.forClass(ExecutionScope.class);
+    verify(eventDispatcherMock, timeout(SEPARATE_EXECUTION_TIMEOUT).times(2))
+        .handleExecution(executionScopeArgumentCaptor.capture());
+    List<ExecutionScope> allValues = executionScopeArgumentCaptor.getAllValues();
+    assertThat(allValues).hasSize(2);
+    assertThat(allValues.get(1).getEvents()).hasSize(2);
+    verify(retryTimerEventSourceMock, never())
+        .scheduleOnce(eq(customResource), eq(GenericRetry.DEFAULT_INITIAL_INTERVAL));
+  }
+
   private void waitMinimalTime() {
     try {
       Thread.sleep(50);