fix: dependent event handling with temp cache not cleared in some corner cases (#1994)

Author: csviri

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java

@@ -128,6 +128,10 @@ protected R handleUpdate(R actual, R desired, P primary, Context<P> context) {
 
   @SuppressWarnings("unused")
   public R create(R target, P primary, Context<P> context) {
+    if (useSSA(context)) {
+      // setting resource version for SSA so only created if it doesn't exist already
+      target.getMetadata().setResourceVersion("1");
+    }
     final var resource = prepare(target, primary, "Creating");
     return useSSA(context)
         ? resource
@@ -138,15 +142,23 @@ public R create(R target, P primary, Context<P> context) {
   }
 
   public R update(R actual, R target, P primary, Context<P> context) {
+    if (log.isDebugEnabled()) {
+      log.debug("Updating actual resource: {} version: {}", ResourceID.fromResource(actual),
+          actual.getMetadata().getResourceVersion());
+    }
+    R updatedResource;
     if (useSSA(context)) {
       target.getMetadata().setResourceVersion(actual.getMetadata().getResourceVersion());
-      return prepare(target, primary, "Updating")
+      updatedResource = prepare(target, primary, "Updating")
           .fieldManager(context.getControllerConfiguration().fieldManager())
           .forceConflicts().serverSideApply();
     } else {
       var updatedActual = updaterMatcher.updateResource(actual, target, context);
-      return prepare(updatedActual, primary, "Updating").replace();
+      updatedResource = prepare(updatedActual, primary, "Updating").replace();
     }
+    log.debug("Resource version after update: {}",
+        updatedResource.getMetadata().getResourceVersion());
+    return updatedResource;
   }
 
   public Result<R> match(R actualResource, P primary, Context<P> context) {

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

@@ -51,7 +51,7 @@
  * </p>
  * <br>
  * <p>
- * 2. Additional API is provided that is ment to be used with the combination of the previous one,
+ * 2. Additional API is provided that is meant to be used with the combination of the previous one,
  * and the goal is to filter out events that are the results of updates and creates made by the
  * controller itself. For example if in reconciler a ConfigMaps is created, there should be an
  * Informer in place to handle change events of that ConfigMap, but since it has bean created (or
@@ -113,9 +113,9 @@ public InformerEventSource(InformerConfiguration<R> configuration, KubernetesCli
   @Override
   public void onAdd(R newResource) {
     if (log.isDebugEnabled()) {
-      log.debug("On add event received for resource id: {} type: {}",
+      log.debug("On add event received for resource id: {} type: {} version: {}",
           ResourceID.fromResource(newResource),
-          resourceType().getSimpleName());
+          resourceType().getSimpleName(), newResource.getMetadata().getResourceVersion());
     }
     primaryToSecondaryIndex.onAddOrUpdate(newResource);
     onAddOrUpdate(Operation.ADD, newResource, null,
@@ -125,9 +125,12 @@ public void onAdd(R newResource) {
   @Override
   public void onUpdate(R oldObject, R newObject) {
     if (log.isDebugEnabled()) {
-      log.debug("On update event received for resource id: {} type: {}",
+      log.debug(
+          "On update event received for resource id: {} type: {} version: {} old version: {} ",
           ResourceID.fromResource(newObject),
-          resourceType().getSimpleName());
+          resourceType().getSimpleName(),
+          newObject.getMetadata().getResourceVersion(),
+          oldObject.getMetadata().getResourceVersion());
     }
     primaryToSecondaryIndex.onAddOrUpdate(newObject);
     onAddOrUpdate(Operation.UPDATE, newObject, oldObject,
@@ -282,17 +285,26 @@ private void handleRecentResourceOperationAndStopEventRecording(Operation operat
         log.debug(
             "Did not found event in buffer with target version and resource id: {}", resourceID);
         temporaryResourceCache.unconditionallyCacheResource(newResource);
-      } else if (eventRecorder.containsEventWithVersionButItsNotLastOne(
-          resourceID, newResource.getMetadata().getResourceVersion())) {
-        R lastEvent = eventRecorder.getLastEvent(resourceID);
-        log.debug(
-            "Found events in event buffer but the target event is not last for id: {}. Propagating event.",
-            resourceID);
-        if (eventAcceptedByFilter(operation, newResource, oldResource)) {
-          propagateEvent(lastEvent);
+      } else {
+        // 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.removeResourceFromCache(newResource);
+        if (eventRecorder.containsEventWithVersionButItsNotLastOne(
+            resourceID, newResource.getMetadata().getResourceVersion())) {
+          R lastEvent = eventRecorder.getLastEvent(resourceID);
+
+          log.debug(
+              "Found events in event buffer but the target event is not last for id: {}. Propagating event.",
+              resourceID);
+          if (eventAcceptedByFilter(operation, newResource, oldResource)) {
+            propagateEvent(lastEvent);
+          }
         }
       }
     } finally {
+      log.debug("Stopping event recording for: {}", resourceID);
       eventRecorder.stopEventRecording(resourceID);
     }
   }