Convinient API to re-schedule events for UpdateControl

Author: csviri

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/UpdateControl.java

@@ -1,12 +1,16 @@
 package io.javaoperatorsdk.operator.api;
 
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+
 import io.fabric8.kubernetes.client.CustomResource;
 
 public class UpdateControl<T extends CustomResource> {
 
   private final T customResource;
   private final boolean updateStatusSubResource;
   private final boolean updateCustomResource;
+  private Long reScheduleDelay = null;
 
   private UpdateControl(
       T customResource, boolean updateStatusSubResource, boolean updateCustomResource) {
@@ -40,6 +44,19 @@ public static <T extends CustomResource> UpdateControl<T> noUpdate() {
     return new UpdateControl<>(null, false, false);
   }
 
+  public UpdateControl withReSchedule(long delay, TimeUnit timeUnit) {
+    return withReSchedule(timeUnit.toMillis(delay));
+  }
+
+  public UpdateControl withReSchedule(long delay) {
+    this.reScheduleDelay = delay;
+    return this;
+  }
+
+  public Optional<Long> getReScheduleDelay() {
+    return Optional.ofNullable(reScheduleDelay);
+  }
+
   public T getCustomResource() {
     return customResource;
   }

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

@@ -163,13 +163,21 @@ void eventProcessingFinished(
         cleanupAfterDeletedEvent(executionScope.getCustomResourceUid());
       } else {
         cacheUpdatedResourceIfChanged(executionScope, postExecutionControl);
+        reScheduleExecutionIfInstructed(postExecutionControl, executionScope.getCustomResource());
         executeBufferedEvents(executionScope.getCustomResourceUid());
       }
     } finally {
       lock.unlock();
     }
   }
 
+  private void reScheduleExecutionIfInstructed(PostExecutionControl<R> postExecutionControl,
+      R customResource) {
+    postExecutionControl.getReScheduleDelay().ifPresent(delay -> eventSourceManager
+        .getRetryTimerEventSource()
+        .scheduleOnce(customResource, delay));
+  }
+
   /**
    * Regarding the events there are 2 approaches we can take. Either retry always when there are new
    * events (received meanwhile retry is in place or already in buffer) instantly or always wait

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

@@ -140,13 +140,20 @@ private PostExecutionControl<R> handleCreateOrUpdate(
       } else if (updateControl.isUpdateCustomResource()) {
         updatedCustomResource = updateCustomResource(updateControl.getCustomResource());
       }
+      return createPostExecutionControl(updatedCustomResource, updateControl);
+    }
+  }
 
-      if (updatedCustomResource != null) {
-        return PostExecutionControl.customResourceUpdated(updatedCustomResource);
-      } else {
-        return PostExecutionControl.defaultDispatch();
-      }
+  private PostExecutionControl<R> createPostExecutionControl(R updatedCustomResource,
+      UpdateControl<R> updateControl) {
+    PostExecutionControl<R> postExecutionControl;
+    if (updatedCustomResource != null) {
+      postExecutionControl = PostExecutionControl.customResourceUpdated(updatedCustomResource);
+    } else {
+      postExecutionControl = PostExecutionControl.defaultDispatch();
     }
+    updateControl.getReScheduleDelay().ifPresent(postExecutionControl::withReSchedule);
+    return postExecutionControl;
   }
 
   private PostExecutionControl<R> handleDelete(R resource, Context<R> context) {

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

@@ -7,11 +7,11 @@
 public final class PostExecutionControl<R extends CustomResource<?, ?>> {
 
   private final boolean onlyFinalizerHandled;
-
   private final R updatedCustomResource;
-
   private final RuntimeException runtimeException;
 
+  private Long reScheduleDelay = null;
+
   private PostExecutionControl(
       boolean onlyFinalizerHandled,
       R updatedCustomResource,
@@ -54,10 +54,19 @@ public boolean exceptionDuringExecution() {
     return runtimeException != null;
   }
 
+  public PostExecutionControl withReSchedule(long delay) {
+    this.reScheduleDelay = delay;
+    return this;
+  }
+
   public Optional<RuntimeException> getRuntimeException() {
     return Optional.ofNullable(runtimeException);
   }
 
+  public Optional<Long> getReScheduleDelay() {
+    return Optional.ofNullable(reScheduleDelay);
+  }
+
   @Override
   public String toString() {
     return "PostExecutionControl{"

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

@@ -23,15 +23,7 @@
 import static io.javaoperatorsdk.operator.TestUtils.testCustomResource;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.any;
-import static org.mockito.Mockito.doAnswer;
-import static org.mockito.Mockito.doCallRealMethod;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.timeout;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.*;
 
 class DefaultEventHandlerTest {
 
@@ -187,7 +179,6 @@ public void successfulExecutionResetsTheRetry() {
     Event event = prepareCREvent();
     TestCustomResource customResource = testCustomResource();
     customResource.getMetadata().setUid(event.getRelatedCustomResourceUid());
-    ExecutionScope executionScope = new ExecutionScope(Arrays.asList(event), customResource, null);
     PostExecutionControl postExecutionControlWithException =
         PostExecutionControl.exceptionDuringExecution(new RuntimeException("test"));
     PostExecutionControl defaultDispatchControl = PostExecutionControl.defaultDispatch();
@@ -222,6 +213,18 @@ public void successfulExecutionResetsTheRetry() {
     assertThat(executionScopes.get(1).getRetryInfo().isLastAttempt()).isEqualTo(false);
   }
 
+  @Test
+  public void scheduleTimedEventIfInstructedByPostExecutionControl() {
+    var testDelay = 10000l;
+    when(eventDispatcherMock.handleExecution(any()))
+        .thenReturn(PostExecutionControl.defaultDispatch().withReSchedule(testDelay));
+
+    defaultEventHandler.handleEvent(prepareCREvent());
+
+    verify(retryTimerEventSourceMock, timeout(SEPARATE_EXECUTION_TIMEOUT).times(1))
+        .scheduleOnce(any(), eq(testDelay));
+  }
+
   private void waitMinimalTime() {
     try {
       Thread.sleep(50);

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

@@ -289,6 +289,20 @@ public boolean isLastAttempt() {
     assertThat(retryInfo.isLastAttempt()).isEqualTo(true);
   }
 
+  @Test
+  void setReScheduleToPostExecutionControlFromUpdateControl() {
+    testCustomResource.addFinalizer(DEFAULT_FINALIZER);
+
+    when(controller.createOrUpdateResource(eq(testCustomResource), any()))
+        .thenReturn(
+            UpdateControl.updateStatusSubResource(testCustomResource).withReSchedule(1000l));
+
+    PostExecutionControl control = eventDispatcher.handleExecution(
+        executionScopeWithCREvent(Watcher.Action.ADDED, testCustomResource));
+
+    assertThat(control.getReScheduleDelay().get()).isEqualTo(1000l);
+  }
+
   private void markForDeletion(CustomResource customResource) {
     customResource.getMetadata().setDeletionTimestamp("2019-8-10");
   }