wip

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/TemporaryResourceCache.java

@@ -127,9 +127,8 @@ private synchronized EventHandling onEvent(
     }
     var cached = cache.get(resourceId);
     EventHandling result = EventHandling.NEW;
-    int comp = 0;
     if (cached != null) {
-      comp = ReconcileUtils.compareResourceVersions(resource, cached);
+      int comp = ReconcileUtils.compareResourceVersions(resource, cached);
       if (comp >= 0 || unknownState) {
         cache.remove(resourceId);
         // we propagate event only for our update or newer other can be discarded since we know we

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

@@ -15,13 +15,13 @@
  */
 package io.javaoperatorsdk.operator.processing.event.source.informer;
 
+import java.time.Duration;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.TestInstance;
 
 import io.fabric8.kubernetes.api.model.ObjectMeta;
 import io.fabric8.kubernetes.api.model.apps.Deployment;
@@ -62,7 +62,6 @@
 import static org.mockito.Mockito.when;
 
 @SuppressWarnings({"rawtypes", "unchecked"})
-@TestInstance(value = TestInstance.Lifecycle.PER_METHOD)
 class InformerEventSourceTest {
 
   private static final String PREV_RESOURCE_VERSION = "0";
@@ -146,12 +145,11 @@ void processEventPropagationWithIncorrectAnnotation() {
 
   @Test
   void propagateEventAndRemoveResourceFromTempCacheIfResourceVersionMismatch() {
+    withRealTemporaryResourceCache();
+
     Deployment cachedDeployment = testDeployment();
     cachedDeployment.getMetadata().setResourceVersion(PREV_RESOURCE_VERSION);
-    when(temporaryResourceCache.getResourceFromCache(any()))
-        .thenReturn(Optional.of(cachedDeployment));
-    when(temporaryResourceCache.onAddOrUpdateEvent(any(), any(), any()))
-        .thenReturn(EventHandling.NEW);
+    temporaryResourceCache.putResource(cachedDeployment);
 
     informerEventSource.onUpdate(cachedDeployment, testDeployment());
 
@@ -247,16 +245,36 @@ void handlesPrevResourceVersionForUpdateInCaseOfMultipleUpdates() {
   }
 
   @Test
-  void doesNotPropagateEventIfReceivedBeforeUpdate() throws InterruptedException {
+  void doesNotPropagateEventIfReceivedBeforeUpdate() {
     withRealTemporaryResourceCache();
 
     CountDownLatch latch = sendForEventFilteringUpdate(2);
     informerEventSource.onUpdate(
         deploymentWithResourceVersion(1), deploymentWithResourceVersion(2));
     latch.countDown();
 
-    Thread.sleep(100);
-    verify(informerEventSource, never()).handleEvent(any(), any(), any(), any());
+    assertNoEventProduced();
+  }
+
+  @Test
+  void filterAddEventBeforeUpdate() {
+    withRealTemporaryResourceCache();
+
+    CountDownLatch latch = sendForEventFilteringUpdate(2);
+    informerEventSource.onAdd(deploymentWithResourceVersion(1));
+    latch.countDown();
+
+    assertNoEventProduced();
+  }
+
+  private void assertNoEventProduced() {
+    await()
+        .pollDelay(Duration.ofMillis(50))
+        .timeout(Duration.ofMillis(51))
+        .untilAsserted(
+            () -> {
+              verify(informerEventSource, never()).handleEvent(any(), any(), any(), any());
+            });
   }
 
   private void expectHandleEvent(int newResourceVersion, int oldResourceVersion) {
@@ -292,7 +310,7 @@ private CountDownLatch sendForEventFilteringUpdate(Deployment deployment, int re
   }
 
   private void withRealTemporaryResourceCache() {
-    temporaryResourceCache = new TemporaryResourceCache<>(true);
+    temporaryResourceCache = spy(new TemporaryResourceCache<>(true));
     informerEventSource.setTemporalResourceCache(temporaryResourceCache);
   }
 

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

@@ -16,12 +16,8 @@
 package io.javaoperatorsdk.operator.processing.event.source.informer;
 
 import java.util.Map;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
 
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
 import io.fabric8.kubernetes.api.model.ConfigMap;
@@ -126,28 +122,72 @@ void nonComparableResourceVersionsDisables() {
         .isEmpty();
   }
 
-  @Disabled("todo")
   @Test
-  void lockedEventBeforePut() throws Exception {
+  void eventReceivedDuringFiltering() throws Exception {
     var testResource = testResource();
 
     temporaryResourceCache.startEventFilteringModify(ResourceID.fromResource(testResource));
 
-    ExecutorService ex = Executors.newSingleThreadExecutor();
-    try {
-      var result =
-          ex.submit(
-              () ->
-                  temporaryResourceCache.onAddOrUpdateEvent(
-                      ResourceAction.ADDED, testResource, null));
-
-      temporaryResourceCache.putResource(testResource);
-      assertThat(result.isDone()).isFalse();
-      temporaryResourceCache.doneEventFilterModify(ResourceID.fromResource(testResource), "3");
-      assertThat(result.get(10, TimeUnit.SECONDS)).isEqualTo(EventHandling.NEW);
-    } finally {
-      ex.shutdownNow();
-    }
+    temporaryResourceCache.putResource(testResource);
+    assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource)))
+        .isPresent();
+
+    temporaryResourceCache.onAddOrUpdateEvent(ResourceAction.ADDED, testResource, null);
+    assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource)))
+        .isEmpty();
+
+    var doneRes =
+        temporaryResourceCache.doneEventFilterModify(ResourceID.fromResource(testResource), "2");
+
+    assertThat(doneRes).isEmpty();
+    assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource)))
+        .isEmpty();
+  }
+
+  @Test
+  void newerEventDuringFiltering() {
+    var testResource = testResource();
+
+    temporaryResourceCache.startEventFilteringModify(ResourceID.fromResource(testResource));
+
+    temporaryResourceCache.putResource(testResource);
+    assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource)))
+        .isPresent();
+
+    var testResource2 = testResource();
+    testResource2.getMetadata().setResourceVersion("3");
+    temporaryResourceCache.onAddOrUpdateEvent(ResourceAction.UPDATED, testResource2, testResource);
+    assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource)))
+        .isEmpty();
+
+    var doneRes =
+        temporaryResourceCache.doneEventFilterModify(ResourceID.fromResource(testResource), "2");
+
+    assertThat(doneRes).isPresent();
+    assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource)))
+        .isEmpty();
+  }
+
+  @Test
+  void eventAfterFiltering() {
+    var testResource = testResource();
+
+    temporaryResourceCache.startEventFilteringModify(ResourceID.fromResource(testResource));
+
+    temporaryResourceCache.putResource(testResource);
+    assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource)))
+        .isPresent();
+
+    var doneRes =
+        temporaryResourceCache.doneEventFilterModify(ResourceID.fromResource(testResource), "2");
+
+    assertThat(doneRes).isEmpty();
+    assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource)))
+        .isPresent();
+
+    temporaryResourceCache.onAddOrUpdateEvent(ResourceAction.ADDED, testResource, null);
+    assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource)))
+        .isEmpty();
   }
 
   @Test