first unit test, minor tweaks

Author: csviri

operator-framework/src/main/java/com/github/containersolutions/operator/Operator.java

@@ -62,8 +62,6 @@ private <R extends CustomResource> void registerController(ResourceController<R>
 
             eventScheduler = new EventScheduler(eventDispatcher);
 
-            eventScheduler.startProcessing();
-
             registerWatches(controller, client, resClass, watchAllNamespaces, targetNamespaces);
         } else {
             throw new OperatorException("CRD '" + resClass.getSimpleName() + "' with version '"

operator-framework/src/main/java/com/github/containersolutions/operator/processing/CustomResourceEvent.java

@@ -12,7 +12,7 @@ class CustomResourceEvent {
 
     private final Watcher.Action action;
     private final CustomResource resource;
-    private Integer retriesIndex = 0;
+    private Integer retryIndex = 0;
 
     CustomResourceEvent(Watcher.Action action, CustomResource resource) {
         this.action = action;
@@ -54,13 +54,17 @@ public Boolean isSameResourceAndNewerVersion(CustomResourceEvent otherEvent) {
 
     }
 
+    public Integer getRetryIndex() {
+        return retryIndex;
+    }
+
     @Override
     public String toString() {
         return "CustomResourceEvent{" +
                 "action=" + action +
                 ", resource=[ name=" + resource.getMetadata().getName() + ", kind=" + resource.getKind() +
                 ", apiVersion=" + resource.getApiVersion() + "]" +
-                ", retriesIndex=" + retriesIndex +
+                ", retriesIndex=" + retryIndex +
                 '}';
     }
 }

operator-framework/src/main/java/com/github/containersolutions/operator/processing/EventDispatcher.java

@@ -40,7 +40,7 @@ public EventDispatcher(ResourceController<R> controller,
         this.k8sClient = k8sClient;
     }
 
-    void handleEvent(Watcher.Action action, R resource) {
+    public void handleEvent(Watcher.Action action, R resource) {
         if (action == Watcher.Action.MODIFIED || action == Watcher.Action.ADDED) {
             // we don't want to call delete resource if it not contains our finalizer,
             // since the resource still can be updates when marked for deletion and contains other finalizers

operator-framework/src/main/java/com/github/containersolutions/operator/processing/EventScheduler.java

@@ -16,7 +16,6 @@
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.ReentrantLock;
 
 
@@ -58,7 +57,6 @@ public class EventScheduler<R extends CustomResource> implements Watcher<R> {
     private final Map<String, ResourceScheduleHolder> eventsScheduledForProcessing = new HashMap<>();
     private final Map<String, CustomResourceEvent> eventsUnderProcessing = new HashMap<>();
 
-    private AtomicBoolean processingEnabled = new AtomicBoolean(false);
     private ReentrantLock lock = new ReentrantLock();
 
     public EventScheduler(EventDispatcher<R> eventDispatcher) {
@@ -71,34 +69,26 @@ public EventScheduler(EventDispatcher<R> eventDispatcher) {
         executor.setRemoveOnCancelPolicy(true);
     }
 
-    public void startProcessing() {
-        processingEnabled.set(true);
-    }
-
     @Override
     public void eventReceived(Watcher.Action action, R resource) {
-        if (!processingEnabled.get()) return;
-
         log.debug("Event received for action: {}, {}: {}", action.toString().toLowerCase(), resource.getClass().getSimpleName(),
                 resource.getMetadata().getName());
-
         CustomResourceEvent event = new CustomResourceEvent(action, resource);
         scheduleEvent(event);
     }
 
-    // todo we want to strictly scheduler resources for execution, so if an event is under processing we should wait
-    //  with the incoming event to be scheduled/executed until that one is not finished
-
-    // todo handle delete event: cleanup when a real delete arrived
-    // todo discuss new version vs new generation comparison
     void scheduleEvent(CustomResourceEvent newEvent) {
         log.debug("Current queue size {}", executor.getQueue().size());
         log.info("Scheduling event: {}", newEvent.getEventInfo());
         try {
             // we have to lock since the fabric8 client event handling is multi-threaded,
             // so in the following part could be a race condition when multiple events are received for same resource.
             lock.lock();
-
+            if (newEvent.getAction() == Action.DELETED) {
+                // this is a tricky situation, do we want to process only events which are marked for deletion?
+                // or just ignore the problem
+                return;
+            }
             // if there is an event waiting for to be scheduled we just replace that.
             if (eventsNotScheduledYet.containsKey(newEvent.resourceUid()) &&
                     newEvent.isSameResourceAndNewerVersion(eventsNotScheduledYet.get(newEvent.resourceUid()))) {
@@ -130,9 +120,11 @@ void scheduleEvent(CustomResourceEvent newEvent) {
                     return;
                 }
             }
+
+            // todo handle backoff instances
             backoffSchedulerCache.put(newEvent, backOff.start());
             ScheduledFuture<?> scheduledTask = executor.schedule(new EventConsumer(newEvent, eventDispatcher, this),
-                    backoffSchedulerCache.get(newEvent).nextBackOff(), TimeUnit.MILLISECONDS);
+                    newEvent.getRetryIndex() < 1 ? 0 : backoffSchedulerCache.get(newEvent).nextBackOff(), TimeUnit.MILLISECONDS);
             eventsScheduledForProcessing.put(newEvent.resourceUid(), new ResourceScheduleHolder(newEvent, scheduledTask));
         } finally {
             lock.unlock();

operator-framework/src/test/java/com/github/containersolutions/operator/EventSchedulerTest.java

@@ -8,37 +8,39 @@
 import io.fabric8.kubernetes.client.Watcher;
 import org.junit.jupiter.api.Test;
 
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.*;
 
 class EventSchedulerTest {
 
     @SuppressWarnings("unchecked")
     private EventDispatcher<CustomResource> eventDispatcher = mock(EventDispatcher.class);
 
+    private EventScheduler<CustomResource> eventScheduler = new EventScheduler(eventDispatcher);
+
     @Test
-    void dontScheduleReceivedEventIfProcessingNotStarted() {
-        EventScheduler<CustomResource> eventScheduler = spy(new EventScheduler<>(eventDispatcher));
+    public void schedulesEvent() {
+        CustomResource resource = sampleResource();
 
-        eventScheduler.eventReceived(any(), any());
+        eventScheduler.eventReceived(Watcher.Action.MODIFIED, resource);
 
-//        verify(eventScheduler, times(0)).scheduleEvent(any());
+        waitMinimalTimeForExecution();
+        verify(eventDispatcher, times(1)).handleEvent(Watcher.Action.MODIFIED, resource);
     }
 
     @Test
-    void scheduleReceivedEventIfProcessingStarted() {
-        EventScheduler<CustomResource> eventScheduler = spy(new EventScheduler<>(eventDispatcher));
-
-        eventScheduler.eventReceived(Watcher.Action.ADDED, getResource());
-        eventScheduler.startProcessing();
-        eventScheduler.eventReceived(Watcher.Action.ADDED, getResource());
+    public void eventsAreNotExecutedConcurrentlyForSameResource() {
 
-//        verify(eventScheduler, times(1)).scheduleEvent(any());
     }
 
+    private void waitMinimalTimeForExecution() {
+        try {
+            Thread.sleep(300);
+        } catch (InterruptedException e) {
+            throw new IllegalStateException(e);
+        }
+    }
 
-    CustomResource getResource() {
+    CustomResource sampleResource() {
         TestCustomResource resource = new TestCustomResource();
         resource.setMetadata(new ObjectMetaBuilder()
                 .withCreationTimestamp("creationTimestamp")