fix: keep track of non existing resource

Author: scrocquesel

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

@@ -26,7 +26,7 @@
 public abstract class CachingEventSource<T, P extends HasMetadata>
     extends AbstractResourceEventSource<P, T> implements Cache<T> {
 
-  protected UpdatableCache<T> cache;
+  protected UpdatableCache<Optional<T>> cache;
 
   public CachingEventSource(Class<T> resourceClass) {
     super(resourceClass);
@@ -35,7 +35,7 @@ public CachingEventSource(Class<T> resourceClass) {
 
   @Override
   public Optional<T> get(ResourceID resourceID) {
-    return cache.get(resourceID);
+    return cache.get(resourceID).orElse(Optional.empty());
   }
 
   @Override
@@ -50,17 +50,24 @@ public Stream<ResourceID> keys() {
 
   @Override
   public Stream<T> list(Predicate<T> predicate) {
-    return cache.list(predicate);
+    return cache.list(c -> c.isPresent() && predicate.test(c.get())).map(c -> c.get());
   }
 
   protected void handleDelete(ResourceID relatedResourceID) {
     if (!isRunning()) {
       return;
     }
+
+
     var cachedValue = cache.get(relatedResourceID);
-    cache.remove(relatedResourceID);
+    cache.put(relatedResourceID, Optional.empty());
     // we only propagate event if the resource was previously in cache
+
     if (cachedValue.isPresent()) {
+      if (cachedValue.get().isPresent()) {
+        getEventHandler().handleEvent(new Event(relatedResourceID));
+      }
+    } else {
       getEventHandler().handleEvent(new Event(relatedResourceID));
     }
   }
@@ -69,19 +76,19 @@ protected void handleEvent(T value, ResourceID relatedResourceID) {
     if (!isRunning()) {
       return;
     }
-    var cachedValue = cache.get(relatedResourceID);
+    var cachedValue = cache.get(relatedResourceID).orElse(Optional.empty());
     if (cachedValue.map(v -> !v.equals(value)).orElse(true)) {
-      cache.put(relatedResourceID, value);
+      cache.put(relatedResourceID, Optional.of(value));
       getEventHandler().handleEvent(new Event(relatedResourceID));
     }
   }
 
-  protected UpdatableCache<T> initCache() {
+  protected UpdatableCache<Optional<T>> initCache() {
     return new MapCache<>();
   }
 
   public Optional<T> getCachedValue(ResourceID resourceID) {
-    return cache.get(resourceID);
+    return cache.get(resourceID).orElse(Optional.empty());
   }
 
   @Override
@@ -91,7 +98,7 @@ public void stop() throws OperatorException {
 
   @Override
   public Optional<T> getAssociated(P primary) {
-    return cache.get(ResourceID.fromResource(primary));
+    return cache.get(ResourceID.fromResource(primary)).orElse(Optional.empty());
   }
 
   protected static class MapCache<T> implements UpdatableCache<T> {

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSource.java

@@ -70,9 +70,10 @@ private Optional<T> getAndCacheResource(ResourceID resourceID) {
     var resource = resourceCache.get(resourceID);
     if (resource.isPresent()) {
       var value = resourceSupplier.getResource(resource.get());
-      value.ifPresent(v -> cache.put(resourceID, v));
+      cache.put(resourceID, value);
       return value;
     }
+
     return Optional.empty();
   }