consolidating the temporary cache updating logic

Author: shawkins

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

@@ -264,22 +264,8 @@ public synchronized void handleRecentResourceCreate(ResourceID resourceID, R res
 
   private void handleRecentCreateOrUpdate(Operation operation, R newResource, R oldResource) {
     primaryToSecondaryIndex.onAddOrUpdate(newResource);
-    ResourceID resourceID = ResourceID.fromResource(newResource);
-    R cachedResource = get(resourceID).orElse(null);
-    if ((oldResource == null && cachedResource == null)
-        || (cachedResource != null && oldResource != null
-            && cachedResource.getMetadata().getResourceVersion()
-                .equals(oldResource.getMetadata().getResourceVersion()))) {
-      log.debug(
-          "Temporarily moving ahead to target version {} for resource id: {}",
-          newResource.getMetadata().getResourceVersion(), resourceID);
-      temporaryResourceCache.unconditionallyCacheResource(newResource);
-    } else if (temporaryResourceCache.removeResourceFromCache(newResource).isPresent()) {
-      // if the resource is not added to the temp cache, it is cleared, since
-      // the cache is cleared by subsequent events after updates, but if those did not receive
-      // the temp cache is still filled at this point with an old resource
-      log.debug("Cleaning temporary cache for resource id: {}", resourceID);
-    }
+    temporaryResourceCache.putResource(newResource, Optional.ofNullable(oldResource)
+        .map(r -> r.getMetadata().getResourceVersion()).orElse(null));
   }
 
   private boolean useSecondaryToPrimaryIndex() {

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

@@ -91,7 +91,7 @@ public void stop() {
   @Override
   public void handleRecentResourceUpdate(ResourceID resourceID, R resource,
       R previousVersionOfResource) {
-    temporaryResourceCache.putUpdatedResource(resource,
+    temporaryResourceCache.putResource(resource,
         previousVersionOfResource.getMetadata().getResourceVersion());
   }
 
@@ -128,8 +128,10 @@ void setTemporalResourceCache(TemporaryResourceCache<R> temporaryResourceCache)
     this.temporaryResourceCache = temporaryResourceCache;
   }
 
+  @Override
   public abstract void addIndexers(Map<String, Function<R, List<String>>> indexers);
 
+  @Override
   public List<R> byIndex(String indexName, String indexKey) {
     return manager().byIndex(indexName, indexKey);
   }

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

@@ -45,36 +45,33 @@ public synchronized Optional<T> removeResourceFromCache(T resource) {
     return Optional.ofNullable(cache.remove(ResourceID.fromResource(resource)));
   }
 
-  public synchronized void unconditionallyCacheResource(T newResource) {
-    putToCache(newResource, null);
-  }
-
   public synchronized void putAddedResource(T newResource) {
-    ResourceID resourceID = ResourceID.fromResource(newResource);
-    if (managedInformerEventSource.get(resourceID).isEmpty()) {
-      log.debug("Putting resource to cache with ID: {}", resourceID);
-      putToCache(newResource, resourceID);
-    } else {
-      log.debug("Won't put resource into cache found already informer cache: {}", resourceID);
-    }
+    putResource(newResource, null);
   }
 
-  public synchronized void putUpdatedResource(T newResource, String previousResourceVersion) {
+  /**
+   * put the item into the cache if the previousResourceVersion matches the current state. If not
+   * the currently cached item is removed.
+   *
+   * @param previousResourceVersion null indicates an add
+   */
+  public synchronized void putResource(T newResource, String previousResourceVersion) {
     var resourceId = ResourceID.fromResource(newResource);
-    var informerCacheResource = managedInformerEventSource.get(resourceId);
-    if (informerCacheResource.isEmpty()) {
-      log.debug("No cached value present for resource: {}", newResource);
-      return;
-    }
-    // if this is not true that means the cache was already updated
-    if (informerCacheResource.get().getMetadata().getResourceVersion()
-        .equals(previousResourceVersion)) {
-      log.debug("Putting resource to temporal cache with id: {}", resourceId);
+    var cachedResource = getResourceFromCache(resourceId)
+        .orElse(managedInformerEventSource.get(resourceId).orElse(null));
+
+    if ((previousResourceVersion == null && cachedResource == null)
+        || (cachedResource != null && previousResourceVersion != null
+            && cachedResource.getMetadata().getResourceVersion()
+                .equals(previousResourceVersion))) {
+      log.debug(
+          "Temporarily moving ahead to target version {} for resource id: {}",
+          newResource.getMetadata().getResourceVersion(), resourceId);
       putToCache(newResource, resourceId);
     } else {
-      // if something is in cache it's surely obsolete now
-      log.debug("Trying to remove an obsolete resource from cache for id: {}", resourceId);
-      cache.remove(resourceId);
+      if (cache.remove(resourceId) != null) {
+        log.debug("Removed an obsolete resource from cache for id: {}", resourceId);
+      }
     }
   }
 

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

@@ -30,7 +30,7 @@ void updateAddsTheResourceIntoCacheIfTheInformerHasThePreviousResourceVersion()
     prevTestResource.getMetadata().setResourceVersion("0");
     when(informerEventSource.get(any())).thenReturn(Optional.of(prevTestResource));
 
-    temporaryResourceCache.putUpdatedResource(testResource, "0");
+    temporaryResourceCache.putResource(testResource, "0");
 
     var cached = temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource));
     assertThat(cached).isPresent();
@@ -43,7 +43,7 @@ void updateNotAddsTheResourceIntoCacheIfTheInformerHasOtherVersion() {
     informerCachedResource.getMetadata().setResourceVersion("x");
     when(informerEventSource.get(any())).thenReturn(Optional.of(informerCachedResource));
 
-    temporaryResourceCache.putUpdatedResource(testResource, "0");
+    temporaryResourceCache.putResource(testResource, "0");
 
     var cached = temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource));
     assertThat(cached).isNotPresent();