delete handling improvements and test improvements

Signed-off-by: Attila Mészáros <a_meszaros@apple.com>

Author: csviri

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSource.java

@@ -122,8 +122,15 @@ public synchronized void onDelete(R resource, boolean deletedFinalStateUnknown)
             log.debug(
                 "On delete event received. deletedFinalStateUnknown: {}", deletedFinalStateUnknown);
           }
+          var resultEvent =
+              temporaryResourceCache.onDeleteEvent(resource, deletedFinalStateUnknown);
+          if (resultEvent.isEmpty()) {
+            return;
+          }
+          if (resultEvent.orElseThrow().getAction() != ResourceAction.DELETED) {
+            log.warn("Non delete event received on onDelete handling. This should not happen.");
+          }
           primaryToSecondaryIndex.onDelete(resource);
-          temporaryResourceCache.onDeleteEvent(resource, deletedFinalStateUnknown);
           if (acceptedByDeleteFilters(resource, deletedFinalStateUnknown)) {
             propagateEvent(resource);
           }
@@ -151,15 +158,15 @@ private synchronized void onAddOrUpdate(ResourceAction action, R newObject, R ol
     primaryToSecondaryIndex.onAddOrUpdate(newObject);
     var resourceID = ResourceID.fromResource(newObject);
 
-    var eventHandling = temporaryResourceCache.onAddOrUpdateEvent(action, newObject, oldObject);
+    var resultEvent = temporaryResourceCache.onAddOrUpdateEvent(action, newObject, oldObject);
 
-    if (eventHandling.isEmpty()) {
+    if (resultEvent.isEmpty()) {
       log.debug("Deferring event propagation");
     } else if (eventAcceptedByFilter(action, newObject, oldObject)) {
       log.debug(
           "Propagating event for {}, resource with same version not result of a our update.",
           action);
-      var event = eventHandling.get();
+      var event = resultEvent.get();
       handleEvent(
           event.getAction(),
           (R) event.getResource().orElseThrow(),

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/TemporaryResourceCache.java

@@ -48,6 +48,15 @@
  *
  * <p>If comparable resource versions are disabled, then this cache is effectively disabled.
  *
+ * <p>Some principles to realize with the current filtering algorithm:
+ *
+ * <ul>
+ *   <li>We propagate events only if we received an event that has the same resourceVersion or newer
+ *       than resource version from update
+ *   <li>The propagated event should correspond to a possible real world scenario - considering also
+ *       ones that could happen is the Informer does a re-list.
+ * </ul>
+ *
  * @param <T> resource to cache.
  */
 public class TemporaryResourceCache<T extends HasMetadata> {

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSourceTest.java

@@ -180,6 +180,7 @@ void handlesPrevResourceVersionForUpdate() {
         deploymentWithResourceVersion(2), deploymentWithResourceVersion(3));
 
     expectHandleAddEvent(3, 1);
+    expectNoActiveUpdates();
   }
 
   @RepeatedTest(REPEAT_COUNT)
@@ -198,6 +199,7 @@ void handlesPrevResourceVersionForUpdateInCaseOfException() {
     latch.countDown();
 
     expectHandleAddEvent(2, 1);
+    expectNoActiveUpdates();
   }
 
   @RepeatedTest(REPEAT_COUNT)
@@ -213,6 +215,7 @@ void handlesPrevResourceVersionForUpdateInCaseOfMultipleUpdates() {
     latch.countDown();
 
     expectHandleAddEvent(4, 2);
+    expectNoActiveUpdates();
   }
 
   @RepeatedTest(REPEAT_COUNT)
@@ -225,6 +228,7 @@ void doesNotPropagateEventIfReceivedBeforeUpdate() {
     latch.countDown();
 
     assertNoEventProduced();
+    expectNoActiveUpdates();
   }
 
   @RepeatedTest(REPEAT_COUNT)
@@ -242,6 +246,7 @@ void multipleCachingFilteringUpdates() {
         deploymentWithResourceVersion(3), deploymentWithResourceVersion(4));
 
     assertNoEventProduced();
+    expectNoActiveUpdates();
   }
 
   @RepeatedTest(REPEAT_COUNT)
@@ -260,6 +265,7 @@ void multipleCachingFilteringUpdates_variant2() {
     latch2.countDown();
 
     assertNoEventProduced();
+    expectNoActiveUpdates();
   }
 
   @RepeatedTest(REPEAT_COUNT)
@@ -278,6 +284,7 @@ void multipleCachingFilteringUpdates_variant3() {
     latch2.countDown();
 
     assertNoEventProduced();
+    expectNoActiveUpdates();
   }
 
   @RepeatedTest(REPEAT_COUNT)
@@ -296,6 +303,7 @@ void multipleCachingFilteringUpdates_variant4() {
     latch2.countDown();
 
     assertNoEventProduced();
+    expectNoActiveUpdates();
   }
 
   @RepeatedTest(REPEAT_COUNT)
@@ -314,6 +322,7 @@ void multipleCachingFilteringUpdates_variant5() {
         deploymentWithResourceVersion(3), deploymentWithResourceVersion(4));
 
     assertNoEventProduced();
+    expectNoActiveUpdates();
   }
 
   @RepeatedTest(REPEAT_COUNT)
@@ -436,6 +445,7 @@ void propagatesIntermediateEventForExternalUpdateDuringFiltering() {
         deploymentWithResourceVersion(4), deploymentWithResourceVersion(5));
 
     expectHandleAddEvent(5, 2);
+    expectNoActiveUpdates();
   }
 
   @RepeatedTest(REPEAT_COUNT)
@@ -464,6 +474,14 @@ void doesNotPropagateIntermediateEventForOurOwnIntermediateUpdate() {
     verify(eventHandlerMock, never()).handleEvent(any());
 
     latch3.countDown();
+    awaitCachedResourceVersion(resourceId, "5");
+    // drain the filter with the event for our own rv 5 — all events are now own,
+    // summary must be empty and no event propagated.
+    informerEventSource.onUpdate(
+        deploymentWithResourceVersion(4), deploymentWithResourceVersion(5));
+
+    assertNoEventProduced();
+    expectNoActiveUpdates();
   }
 
   @RepeatedTest(REPEAT_COUNT)
@@ -481,6 +499,7 @@ void deleteEventPropagatedIfItWasTheLastEvent() {
     latch.countDown();
 
     expectHandleDeleteEvent(5);
+    expectNoActiveUpdates();
   }
 
   private void awaitCachedResourceVersion(ResourceID resourceId, String resourceVersion) {
@@ -501,6 +520,12 @@ private void assertNoEventProduced() {
         .untilAsserted(() -> verify(informerEventSource, never()).propagateEvent(any()));
   }
 
+  private void expectNoActiveUpdates() {
+    await()
+        .atMost(Duration.ofSeconds(1))
+        .untilAsserted(() -> assertThat(temporaryResourceCache.getActiveUpdates()).isEmpty());
+  }
+
   private void expectHandleAddEvent(int newResourceVersion) {
     await()
         .atMost(Duration.ofSeconds(1))