ghost clear concurrency tests

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

@@ -231,7 +231,6 @@ private String getLastSyncResourceVersion(String namespace) {
     return managedInformerEventSource.manager().lastSyncResourceVersion(namespace);
   }
 
-  // todo tests with combination of event processing
   /**
    * There are (probably extremely rare) circumstances, when we can miss a delete event related to a
    * resources: when we create a resource that is deleted right after by third party and the related

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

@@ -19,6 +19,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 
 import org.junit.jupiter.api.BeforeEach;
@@ -339,6 +340,116 @@ void multipleCachingFilteringUpdates_variant4() {
     assertNoEventProduced();
   }
 
+  @Test
+  void ghostCheckRemovesCachedResourceDuringFilteringUpdate() {
+    var mes = mock(ManagedInformerEventSource.class);
+    var mim = mock(InformerManager.class);
+    when(mes.manager()).thenReturn(mim);
+    when(mim.isWatchingNamespace(any())).thenReturn(true);
+    when(mim.lastSyncResourceVersion(any())).thenReturn("1");
+    when(mim.get(any())).thenReturn(Optional.empty());
+
+    var ghostCheckExecutor = Executors.newScheduledThreadPool(1);
+    temporaryResourceCache = spy(new TemporaryResourceCache<>(true, 50, ghostCheckExecutor, mes));
+    informerEventSource.setTemporalResourceCache(temporaryResourceCache);
+
+    // put resource in cache and start a filtering update
+    var deployment = deploymentWithResourceVersion(2);
+    temporaryResourceCache.putResource(deployment);
+    var resourceId = ResourceID.fromResource(deployment);
+    temporaryResourceCache.startEventFilteringModify(resourceId);
+
+    // advance sync version so ghost check considers the cached resource outdated
+    when(mim.lastSyncResourceVersion(any())).thenReturn("3");
+
+    // ghost check should remove the cached resource
+    await()
+        .untilAsserted(
+            () -> assertThat(temporaryResourceCache.getResourceFromCache(resourceId)).isEmpty());
+
+    // complete the filtering update - the resource should not reappear
+    temporaryResourceCache.doneEventFilterModify(resourceId, "2");
+    assertThat(temporaryResourceCache.getResourceFromCache(resourceId)).isEmpty();
+
+    ghostCheckExecutor.shutdownNow();
+  }
+
+  @Test
+  void ghostCheckRunsConcurrentlyWithPutResource() {
+    var mes = mock(ManagedInformerEventSource.class);
+    var mim = mock(InformerManager.class);
+    when(mes.manager()).thenReturn(mim);
+    when(mim.isWatchingNamespace(any())).thenReturn(true);
+    when(mim.lastSyncResourceVersion(any())).thenReturn("1");
+    when(mim.get(any())).thenReturn(Optional.empty());
+
+    var ghostCheckExecutor = Executors.newScheduledThreadPool(1);
+    temporaryResourceCache = spy(new TemporaryResourceCache<>(true, 50, ghostCheckExecutor, mes));
+    informerEventSource.setTemporalResourceCache(temporaryResourceCache);
+
+    // put a resource that will become a ghost
+    var deployment = deploymentWithResourceVersion(2);
+    temporaryResourceCache.putResource(deployment);
+
+    // advance sync version so ghost check removes it
+    when(mim.lastSyncResourceVersion(any())).thenReturn("3");
+
+    await()
+        .untilAsserted(
+            () ->
+                assertThat(
+                        temporaryResourceCache.getResourceFromCache(
+                            ResourceID.fromResource(deployment)))
+                    .isEmpty());
+
+    // now put a newer resource - should succeed even after ghost removal
+    var newerDeployment = deploymentWithResourceVersion(4);
+    temporaryResourceCache.putResource(newerDeployment);
+    assertThat(
+            temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(newerDeployment)))
+        .isPresent();
+
+    ghostCheckExecutor.shutdownNow();
+  }
+
+  @Test
+  void filteringUpdateAndGhostCheckWithNamespaceChange() {
+    var mes = mock(ManagedInformerEventSource.class);
+    var mim = mock(InformerManager.class);
+    when(mes.manager()).thenReturn(mim);
+    when(mim.isWatchingNamespace(any())).thenReturn(true);
+    when(mim.lastSyncResourceVersion(any())).thenReturn("1");
+    when(mim.get(any())).thenReturn(Optional.empty());
+
+    var ghostCheckExecutor = Executors.newScheduledThreadPool(1);
+    temporaryResourceCache = spy(new TemporaryResourceCache<>(true, 50, ghostCheckExecutor, mes));
+    informerEventSource.setTemporalResourceCache(temporaryResourceCache);
+
+    // start filtering update and put resource
+    var deployment = deploymentWithResourceVersion(2);
+    var resourceId = ResourceID.fromResource(deployment);
+    temporaryResourceCache.startEventFilteringModify(resourceId);
+    temporaryResourceCache.putResource(deployment);
+
+    // namespace becomes unwatched - ghost check should clean up
+    when(mim.isWatchingNamespace(any())).thenReturn(false);
+
+    await()
+        .untilAsserted(
+            () -> assertThat(temporaryResourceCache.getResourceFromCache(resourceId)).isEmpty());
+
+    // complete the filtering update
+    var doneResult = temporaryResourceCache.doneEventFilterModify(resourceId, "2");
+    // resource was already cleaned by ghost check, so no deferred event
+    assertThat(doneResult).isEmpty();
+
+    // put should be rejected since namespace is no longer watched
+    temporaryResourceCache.putResource(deploymentWithResourceVersion(3));
+    assertThat(temporaryResourceCache.getResourceFromCache(resourceId)).isEmpty();
+
+    ghostCheckExecutor.shutdownNow();
+  }
+
   private void assertNoEventProduced() {
     await()
         .pollDelay(Duration.ofMillis(50))