changed scheduling algorithm

Author: csviri

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

@@ -32,16 +32,7 @@ public String resourceUid() {
         return resource.getMetadata().getUid();
     }
 
-    public Boolean sameResourceAs(CustomResourceEvent otherEvent) {
-        return getResource().getMetadata().getUid().equals(otherEvent.getResource().getMetadata().getUid());
-    }
-
-    public Boolean isSameResourceAndNewerVersion(CustomResourceEvent otherEvent) {
-        return sameResourceAs(otherEvent) &&
-                Long.parseLong(getResource().getMetadata().getResourceVersion()) >
-                        Long.parseLong(otherEvent.getResource().getMetadata().getResourceVersion());
 
-    }
 
     public Optional<Long> nextBackOff() {
         retryCount++;

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

@@ -83,24 +83,12 @@ void scheduleEvent(CustomResourceEvent event) {
                 log.debug("Skipping delete event since deletion timestamp is present on resource, so finalizer was in place.");
                 return;
             }
-            if (eventStore.receivedMoreRecentEventBefore(event)) {
-                log.debug("Skipping event processing since was processed event with newer version before. {}", event);
-                return;
-            }
-            eventStore.updateLatestResourceVersionReceived(event);
-
-            if (eventStore.containsOlderVersionOfNotScheduledEvent(event)) {
-                log.debug("Replacing event which is not scheduled yet, since incoming event is more recent. new Event:{}", event);
-                eventStore.addOrReplaceEventAsNotScheduledYet(event);
-                return;
-            }
-            if (eventStore.containsOlderVersionOfEventUnderProcessing(event)) {
+            if (eventStore.containsNotScheduledEvent(event.resourceUid())) {
                 log.debug("Scheduling event for later processing since there is an event under processing for same kind." +
                         " New event: {}", event);
                 eventStore.addOrReplaceEventAsNotScheduledYet(event);
                 return;
             }
-
             Optional<Long> nextBackOff = event.nextBackOff();
             if (!nextBackOff.isPresent()) {
                 log.warn("Event max retry limit reached. Will be discarded. {}", event);
@@ -120,9 +108,8 @@ void eventProcessingFinishedSuccessfully(CustomResourceEvent event) {
         try {
             lock.lock();
             eventStore.removeEventUnderProcessing(event.resourceUid());
-            CustomResourceEvent notScheduledYetEvent = eventStore.removeEventNotScheduledYet(event.resourceUid());
-            if (notScheduledYetEvent != null) {
-                scheduleEvent(notScheduledYetEvent);
+            if (eventStore.containsNotScheduledEvent(event.resourceUid())) {
+                scheduleEvent(eventStore.removeEventNotScheduledYet(event.resourceUid()));
             }
         } finally {
             lock.unlock();
@@ -133,14 +120,8 @@ void eventProcessingFailed(CustomResourceEvent event) {
         try {
             lock.lock();
             eventStore.removeEventUnderProcessing(event.resourceUid());
-            CustomResourceEvent notScheduledYetEvent = eventStore.removeEventNotScheduledYet(event.resourceUid());
-            if (notScheduledYetEvent != null) {
-                if (!notScheduledYetEvent.isSameResourceAndNewerVersion(event)) {
-                    log.warn("The not yet scheduled event has older version then actual event. This is probably a bug.");
-                }
-                // this is the case when we failed processing an event but we already received a new one.
-                // Since since we process declarative resources it correct to schedule the new event.
-                scheduleEvent(notScheduledYetEvent);
+            if (eventStore.containsNotScheduledEvent(event.resourceUid())) {
+                scheduleEvent(eventStore.removeEventNotScheduledYet(event.resourceUid()));
             } else {
                 scheduleEvent(event);
             }

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

@@ -10,13 +10,11 @@ public class EventStore {
 
     private final static Logger log = LoggerFactory.getLogger(EventStore.class);
 
-    private final Map<String, Long> lastResourceVersion = new HashMap<>();
     private final Map<String, CustomResourceEvent> eventsNotScheduledYet = new HashMap<>();
     private final Map<String, CustomResourceEvent> eventsUnderProcessing = new HashMap<>();
 
-    public boolean containsOlderVersionOfNotScheduledEvent(CustomResourceEvent newEvent) {
-        return eventsNotScheduledYet.containsKey(newEvent.resourceUid()) &&
-                newEvent.isSameResourceAndNewerVersion(eventsNotScheduledYet.get(newEvent.resourceUid()));
+    public boolean containsNotScheduledEvent(String uuid) {
+        return eventsNotScheduledYet.containsKey(uuid);
     }
 
     public CustomResourceEvent removeEventNotScheduledYet(String uid) {
@@ -27,38 +25,11 @@ public void addOrReplaceEventAsNotScheduledYet(CustomResourceEvent event) {
         eventsNotScheduledYet.put(event.resourceUid(), event);
     }
 
-    public boolean containsOlderVersionOfEventUnderProcessing(CustomResourceEvent newEvent) {
-        return eventsUnderProcessing.containsKey(newEvent.resourceUid()) &&
-                newEvent.isSameResourceAndNewerVersion(eventsUnderProcessing.get(newEvent.resourceUid()));
-    }
-
-
     public void addEventUnderProcessing(CustomResourceEvent event) {
         eventsUnderProcessing.put(event.resourceUid(), event);
     }
-
     public CustomResourceEvent removeEventUnderProcessing(String uid) {
         return eventsUnderProcessing.remove(uid);
     }
 
-    public void updateLatestResourceVersionReceived(CustomResourceEvent event) {
-        Long current = lastResourceVersion.get(event.resourceUid());
-        long received = Long.parseLong(event.getResource().getMetadata().getResourceVersion());
-        if (current == null || received > current) {
-            lastResourceVersion.put(event.resourceUid(), received);
-            log.debug("Resource version for {} updated from {} to {}", event.getResource().getMetadata().getName(), current, received);
-        } else {
-            log.debug("Resource version for {} not updated from {}", event.getResource().getMetadata().getName(), current);
-        }
-    }
-
-    public boolean receivedMoreRecentEventBefore(CustomResourceEvent customResourceEvent) {
-        Long lastVersionProcessed = lastResourceVersion.get(customResourceEvent.resourceUid());
-        if (lastVersionProcessed == null) {
-            return false;
-        } else {
-            return lastVersionProcessed > Long.parseLong(customResourceEvent.getResource()
-                    .getMetadata().getResourceVersion());
-        }
-    }
 }