simplified scheduling algorithm

Author: csviri

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

@@ -20,10 +20,6 @@ class EventConsumer implements Runnable {
 
     @Override
     public void run() {
-        boolean stillScheduledForProcessing = eventScheduler.eventProcessingStarted(event);
-        if (!stillScheduledForProcessing) {
-            return;
-        }
         if (processEvent()) {
             eventScheduler.eventProcessingFinishedSuccessfully(event);
         } else {

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

@@ -9,7 +9,6 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.Optional;
-import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
@@ -100,47 +99,17 @@ void scheduleEvent(CustomResourceEvent event) {
                 eventStore.addOrReplaceEventAsNotScheduledYet(event);
                 return;
             }
-            if (eventStore.containsEventScheduledForProcessing(event.resourceUid())) {
-                EventStore.ResourceScheduleHolder scheduleHolder = eventStore.getEventScheduledForProcessing(event.resourceUid());
-                CustomResourceEvent scheduledEvent = scheduleHolder.getCustomResourceEvent();
-                ScheduledFuture<?> scheduledFuture = scheduleHolder.getScheduledFuture();
-                // If newEvent is newer than existing in queue, cancel and remove queuedEvent
-                if (event.isSameResourceAndNewerVersion(scheduledEvent)) {
-                    log.debug("Scheduled event canceled because incoming event is newer. Discarded: {}, New: {}", scheduledEvent, event);
-                    scheduledFuture.cancel(false);
-                    eventStore.removeEventScheduledForProcessingVersionAware(scheduledEvent.resourceUid());
-                }
-            }
-
             Optional<Long> nextBackOff = event.nextBackOff();
             if (!nextBackOff.isPresent()) {
                 log.warn("Event limited max retry limit ({}), will be discarded. {}", MAX_RETRY_COUNT, event);
                 return;
             }
             log.debug("Creating scheduled task for event: {}", event);
-            ScheduledFuture<?> scheduledTask = executor.schedule(new EventConsumer(event, eventDispatcher, this),
+            executor.schedule(new EventConsumer(event, eventDispatcher, this),
                     nextBackOff.get(), TimeUnit.MILLISECONDS);
-            eventStore.addEventScheduledForProcessing(new EventStore.ResourceScheduleHolder(event, scheduledTask));
-        } finally {
-            log.info("Scheduling event finished: {}", event);
-            lock.unlock();
-        }
-    }
-
-    boolean eventProcessingStarted(CustomResourceEvent event) {
-        try {
-            lock.lock();
-            EventStore.ResourceScheduleHolder res = eventStore.removeEventScheduledForProcessingVersionAware(event);
-            if (res == null) {
-                // Double checking if the event is still scheduled. This is a corner case, but can actually happen.
-                // In detail: it can happen that we scheduled an event for processing, it took some time that is was picked
-                // by executor, and it was removed during that time from the schedule but not cancelled yet. So to be correct
-                // this should be checked also here. In other word scheduleEvent function can run in parallel with eventDispatcher.
-                return false;
-            }
             eventStore.addEventUnderProcessing(event);
-            return true;
         } finally {
+            log.info("Scheduling event finished: {}", event);
             lock.unlock();
         }
     }

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

@@ -5,7 +5,6 @@
 
 import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.ScheduledFuture;
 
 public class EventStore {
 
@@ -14,7 +13,6 @@ public class EventStore {
     private final Map<String, Long> lastResourceVersion = new HashMap<>();
 
     private final Map<String, CustomResourceEvent> eventsNotScheduledYet = new HashMap<>();
-    private final Map<String, ResourceScheduleHolder> eventsScheduledForProcessing = new HashMap<>();
     private final Map<String, CustomResourceEvent> eventsUnderProcessing = new HashMap<>();
 
     public boolean containsOlderVersionOfNotScheduledEvent(CustomResourceEvent newEvent) {
@@ -35,35 +33,11 @@ public boolean containsOlderVersionOfEventUnderProcessing(CustomResourceEvent ne
                 newEvent.isSameResourceAndNewerVersion(eventsUnderProcessing.get(newEvent.resourceUid()));
     }
 
-    public boolean containsEventScheduledForProcessing(String uid) {
-        return eventsScheduledForProcessing.containsKey(uid);
-    }
 
     public void addEventUnderProcessing(CustomResourceEvent event) {
         eventsUnderProcessing.put(event.resourceUid(), event);
     }
 
-    public ResourceScheduleHolder getEventScheduledForProcessing(String uid) {
-        return eventsScheduledForProcessing.get(uid);
-    }
-
-    public ResourceScheduleHolder removeEventScheduledForProcessingVersionAware(String uid) {
-        return eventsScheduledForProcessing.remove(uid);
-    }
-
-    public ResourceScheduleHolder removeEventScheduledForProcessingVersionAware(CustomResourceEvent customResourceEvent) {
-        ResourceScheduleHolder holder = eventsScheduledForProcessing.get(customResourceEvent.resourceUid());
-        if (holder.getCustomResourceEvent().getResource().getMetadata().getResourceVersion().equals(customResourceEvent.getResource().getMetadata().getResourceVersion())) {
-            return removeEventScheduledForProcessingVersionAware(customResourceEvent.resourceUid());
-        } else {
-            return null;
-        }
-    }
-
-    public void addEventScheduledForProcessing(ResourceScheduleHolder resourceScheduleHolder) {
-        eventsScheduledForProcessing.put(resourceScheduleHolder.getCustomResourceEvent().resourceUid(), resourceScheduleHolder);
-    }
-
     public CustomResourceEvent removeEventUnderProcessing(String uid) {
         return eventsUnderProcessing.remove(uid);
     }
@@ -88,22 +62,4 @@ public boolean processedNewerVersionBefore(CustomResourceEvent customResourceEve
                     .getMetadata().getResourceVersion());
         }
     }
-
-    public static class ResourceScheduleHolder {
-        private CustomResourceEvent customResourceEvent;
-        private ScheduledFuture<?> scheduledFuture;
-
-        public ResourceScheduleHolder(CustomResourceEvent customResourceEvent, ScheduledFuture<?> scheduledFuture) {
-            this.customResourceEvent = customResourceEvent;
-            this.scheduledFuture = scheduledFuture;
-        }
-
-        public CustomResourceEvent getCustomResourceEvent() {
-            return customResourceEvent;
-        }
-
-        public ScheduledFuture<?> getScheduledFuture() {
-            return scheduledFuture;
-        }
-    }
 }