fix: event filtering edge case with no-op updates - extended

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

@@ -30,6 +30,7 @@
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
 import io.javaoperatorsdk.operator.api.reconciler.Ignore;
+import io.javaoperatorsdk.operator.api.reconciler.ResourceOperations;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.GarbageCollected;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.ConfiguredDependentResource;
 import io.javaoperatorsdk.operator.processing.GroupVersionKind;
@@ -90,10 +91,12 @@ public R create(R desired, P primary, Context<P> context) {
         desired.getClass(),
         ResourceID.fromResource(desired),
         ssa);
-
     return ssa
         ? context.resourceOperations().serverSideApply(desired, eventSource().orElse(null))
-        : context.resourceOperations().create(desired, eventSource().orElse(null));
+        : context
+            .resourceOperations()
+            .create(
+                desired, eventSource().orElse(null), ResourceOperations.Options.forcedFiltering());
   }
 
   public R update(R actual, R desired, P primary, Context<P> context) {
@@ -114,7 +117,12 @@ public R update(R actual, R desired, P primary, Context<P> context) {
         ssa);
     if (ssa) {
       updatedResource =
-          context.resourceOperations().serverSideApply(desired, eventSource().orElse(null));
+          context
+              .resourceOperations()
+              .serverSideApply(
+                  desired,
+                  eventSource().orElse(null),
+                  ResourceOperations.Options.forcedFiltering());
     } else {
       var updatedActual = GenericResourceUpdater.updateResource(actual, desired, context);
       updatedResource =

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

@@ -138,8 +138,7 @@ private PostExecutionControl<P> handleReconcile(
       } else {
         updatedResource = context.resourceOperations().addFinalizer();
       }
-      return PostExecutionControl.onlyFinalizerAdded(updatedResource)
-          .withReSchedule(BaseControl.INSTANT_RESCHEDULE);
+      return PostExecutionControl.onlyFinalizerAdded(updatedResource);
     } else {
       try {
         return reconcileExecution(executionScope, resourceForExecution, originalResource, context);

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

@@ -39,6 +39,7 @@
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.config.Informable;
 import io.javaoperatorsdk.operator.api.config.NamespaceChangeable;
+import io.javaoperatorsdk.operator.api.reconciler.Experimental;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.RecentOperationCacheFiller;
 import io.javaoperatorsdk.operator.health.InformerHealthIndicator;
 import io.javaoperatorsdk.operator.health.InformerWrappingEventSourceHealthIndicator;
@@ -48,6 +49,8 @@
 import io.javaoperatorsdk.operator.processing.event.source.*;
 import io.javaoperatorsdk.operator.processing.event.source.ResourceAction;
 
+import static io.javaoperatorsdk.operator.api.reconciler.Experimental.API_MIGHT_CHANGE;
+
 @SuppressWarnings("rawtypes")
 public abstract class ManagedInformerEventSource<
         R extends HasMetadata, P extends HasMetadata, C extends Informable<R>>
@@ -88,13 +91,25 @@ public void changeNamespaces(Set<String> namespaces) {
     }
   }
 
+  @Experimental(API_MIGHT_CHANGE)
+  public R eventFilteringUpdateAndCacheResource(R resourceToUpdate, UnaryOperator<R> updateMethod) {
+    return eventFilteringUpdateAndCacheResource(resourceToUpdate, updateMethod, false);
+  }
+
   /**
    * Updates the resource and makes sure that the response is available for the next reconciliation.
    * Also makes sure that the even produced by this update is filtered, thus does not trigger the
    * reconciliation.
    */
+  @Experimental(API_MIGHT_CHANGE)
   @SuppressWarnings("unchecked")
-  public R eventFilteringUpdateAndCacheResource(R resourceToUpdate, UnaryOperator<R> updateMethod) {
+  public R eventFilteringUpdateAndCacheResource(
+      R resourceToUpdate, UnaryOperator<R> updateMethod, boolean forceUpdateFilter) {
+    if (resourceToUpdate.getMetadata().getResourceVersion() == null && !forceUpdateFilter) {
+      log.debug("No resourceVersion set. Skipping event filtering.");
+      return updateAndCacheResource(resourceToUpdate, updateMethod);
+    }
+
     ResourceID id = ResourceID.fromResource(resourceToUpdate);
     log.debug("Starting event filtering and caching update for id={}", id);
     R updatedResource = null;
@@ -134,6 +149,13 @@ public R eventFilteringUpdateAndCacheResource(R resourceToUpdate, UnaryOperator<
     }
   }
 
+  @Experimental(API_MIGHT_CHANGE)
+  public R updateAndCacheResource(R resourceToUpdate, UnaryOperator<R> updateOperation) {
+    var result = updateOperation.apply(resourceToUpdate);
+    handleRecentResourceUpdate(ResourceID.fromResource(resourceToUpdate), result, resourceToUpdate);
+    return result;
+  }
+
   protected abstract void handleEvent(
       ResourceAction action,
       R resource,

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/reconciler/ResourceOperationsTest.java

@@ -84,7 +84,7 @@ void addsFinalizer() {
 
     // Mock successful finalizer addition
     when(controllerEventSource.eventFilteringUpdateAndCacheResource(
-            any(), any(UnaryOperator.class)))
+            any(), any(UnaryOperator.class), anyBoolean()))
         .thenAnswer(
             invocation -> {
               var res = TestUtils.testCustomResource1();
@@ -99,7 +99,7 @@ void addsFinalizer() {
     assertThat(result.hasFinalizer(FINALIZER_NAME)).isTrue();
     assertThat(result.getMetadata().getResourceVersion()).isEqualTo("2");
     verify(controllerEventSource, times(1))
-        .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class));
+        .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean());
   }
 
   @Test
@@ -111,7 +111,7 @@ void addsFinalizerWithSSA() {
 
     // Mock successful SSA finalizer addition
     when(controllerEventSource.eventFilteringUpdateAndCacheResource(
-            any(), any(UnaryOperator.class)))
+            any(), any(UnaryOperator.class), anyBoolean()))
         .thenAnswer(
             invocation -> {
               var res = TestUtils.testCustomResource1();
@@ -126,7 +126,7 @@ void addsFinalizerWithSSA() {
     assertThat(result.hasFinalizer(FINALIZER_NAME)).isTrue();
     assertThat(result.getMetadata().getResourceVersion()).isEqualTo("2");
     verify(controllerEventSource, times(1))
-        .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class));
+        .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean());
   }
 
   @Test
@@ -139,7 +139,7 @@ void removesFinalizer() {
 
     // Mock successful finalizer removal
     when(controllerEventSource.eventFilteringUpdateAndCacheResource(
-            any(), any(UnaryOperator.class)))
+            any(), any(UnaryOperator.class), anyBoolean()))
         .thenAnswer(
             invocation -> {
               var res = TestUtils.testCustomResource1();
@@ -154,7 +154,7 @@ void removesFinalizer() {
     assertThat(result.hasFinalizer(FINALIZER_NAME)).isFalse();
     assertThat(result.getMetadata().getResourceVersion()).isEqualTo("2");
     verify(controllerEventSource, times(1))
-        .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class));
+        .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean());
   }
 
   @Test
@@ -166,7 +166,7 @@ void retriesAddingFinalizerWithoutSSA() {
 
     // First call throws conflict, second succeeds
     when(controllerEventSource.eventFilteringUpdateAndCacheResource(
-            any(), any(UnaryOperator.class)))
+            any(), any(UnaryOperator.class), anyBoolean()))
         .thenThrow(new KubernetesClientException("Conflict", 409, null))
         .thenAnswer(
             invocation -> {
@@ -184,7 +184,7 @@ void retriesAddingFinalizerWithoutSSA() {
     assertThat(result).isNotNull();
     assertThat(result.hasFinalizer(FINALIZER_NAME)).isTrue();
     verify(controllerEventSource, times(2))
-        .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class));
+        .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean());
     verify(resourceOp, times(1)).get();
   }
 
@@ -198,7 +198,7 @@ void nullResourceIsGracefullyHandledOnFinalizerRemovalRetry() {
 
     // First call throws conflict
     when(controllerEventSource.eventFilteringUpdateAndCacheResource(
-            any(), any(UnaryOperator.class)))
+            any(), any(UnaryOperator.class), anyBoolean()))
         .thenThrow(new KubernetesClientException("Conflict", 409, null));
 
     // Return null on retry (resource was deleted)
@@ -207,7 +207,7 @@ void nullResourceIsGracefullyHandledOnFinalizerRemovalRetry() {
     resourceOperations.removeFinalizer(FINALIZER_NAME);
 
     verify(controllerEventSource, times(1))
-        .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class));
+        .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean());
     verify(resourceOp, times(1)).get();
   }
 
@@ -221,7 +221,7 @@ void retriesFinalizerRemovalWithFreshResource() {
 
     // First call throws unprocessable (422), second succeeds
     when(controllerEventSource.eventFilteringUpdateAndCacheResource(
-            any(), any(UnaryOperator.class)))
+            any(), any(UnaryOperator.class), anyBoolean()))
         .thenThrow(new KubernetesClientException("Unprocessable", 422, null))
         .thenAnswer(
             invocation -> {
@@ -243,7 +243,7 @@ void retriesFinalizerRemovalWithFreshResource() {
     assertThat(result.getMetadata().getResourceVersion()).isEqualTo("3");
     assertThat(result.hasFinalizer(FINALIZER_NAME)).isFalse();
     verify(controllerEventSource, times(2))
-        .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class));
+        .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean());
     verify(resourceOp, times(1)).get();
   }
 
@@ -261,15 +261,16 @@ void resourcePatchWithSingleEventSource() {
     when(context.eventSourceRetriever()).thenReturn(eventSourceRetriever);
     when(eventSourceRetriever.getEventSourcesFor(TestCustomResource.class))
         .thenReturn(List.of(managedEventSource));
-    when(managedEventSource.eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class)))
+    when(managedEventSource.eventFilteringUpdateAndCacheResource(
+            any(), any(UnaryOperator.class), anyBoolean()))
         .thenReturn(updatedResource);
 
     var result = resourceOperations.resourcePatch(resource, UnaryOperator.identity());
 
     assertThat(result).isNotNull();
     assertThat(result.getMetadata().getResourceVersion()).isEqualTo("2");
     verify(managedEventSource, times(1))
-        .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class));
+        .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean());
   }
 
   @Test
@@ -303,7 +304,7 @@ void resourcePatchUsesFirstEventSourceIfMultipleEventSourcesPresent() {
     resourceOperations.resourcePatch(resource, UnaryOperator.identity());
 
     verify(eventSource1, times(1))
-        .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class));
+        .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean());
   }
 
   @Test

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

@@ -780,6 +780,47 @@ void filteringUpdateFallsBackToMapperWhenNoPrimaryToSecondaryIndex() {
     verify(eventHandlerMock, times(1)).handleEvent(any());
   }
 
+  @Test
+  void skipsEventFilteringWhenResourceVersionIsNull() {
+    // Without a resourceVersion there is nothing to correlate own-write echoes against, so
+    // event filtering is bypassed: the update is applied and cached directly, no filter window
+    // is opened and no event is propagated through handleEvent.
+    var resourceToUpdate = testDeployment();
+    resourceToUpdate.getMetadata().setResourceVersion(null);
+    var updated = deploymentWithResourceVersion(3);
+
+    var result =
+        informerEventSource.eventFilteringUpdateAndCacheResource(resourceToUpdate, r -> updated);
+
+    assertThat(result).isSameAs(updated);
+    verify(temporaryResourceCache, times(1)).putResource(updated);
+    verify(temporaryResourceCache, never()).startEventFilteringModify(any());
+    verify(temporaryResourceCache, never()).doneEventFilterModify(any());
+    verify(eventHandlerMock, never()).handleEvent(any());
+  }
+
+  @Test
+  void forceUpdateFilterOpensFilterWindowEvenWhenResourceVersionIsNull() {
+    // A write without a resourceVersion (e.g. an SSA finalizer add, which uses no optimistic
+    // locking) would normally skip event filtering. forceUpdateFilter=true forces filtering
+    // anyway so the resulting own event is correlated and does not trigger a spurious
+    // reconciliation.
+    var resourceToUpdate = testDeployment();
+    resourceToUpdate.getMetadata().setResourceVersion(null);
+    var updated = deploymentWithResourceVersion(3);
+    when(temporaryResourceCache.doneEventFilterModify(any())).thenReturn(Optional.empty());
+
+    var result =
+        informerEventSource.eventFilteringUpdateAndCacheResource(
+            resourceToUpdate, r -> updated, true);
+
+    assertThat(result).isSameAs(updated);
+    verify(temporaryResourceCache, times(1)).startEventFilteringModify(any());
+    verify(temporaryResourceCache, times(1)).doneEventFilterModify(any());
+    verify(temporaryResourceCache, times(1)).putResource(updated);
+    verify(eventHandlerMock, never()).handleEvent(any());
+  }
+
   private PrimaryToSecondaryIndex<Deployment> injectIndexMock() throws Exception {
     @SuppressWarnings("unchecked")
     PrimaryToSecondaryIndex<Deployment> indexMock = mock(PrimaryToSecondaryIndex.class);

operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/changenamespace/ChangeNamespaceTestReconciler.java

@@ -53,7 +53,9 @@ public UpdateControl<ChangeNamespaceTestCustomResource> reconcile(
       ChangeNamespaceTestCustomResource primary,
       Context<ChangeNamespaceTestCustomResource> context) {
 
-    context.resourceOperations().serverSideApply(configMap(primary));
+    context
+        .resourceOperations()
+        .serverSideApply(configMap(primary), ResourceOperations.Options.forcedFiltering());
 
     if (primary.getStatus() == null) {
       primary.setStatus(new ChangeNamespaceTestCustomResourceStatus());

operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/onrelistfilter/OnRelistFilterReconciler.java

@@ -34,6 +34,7 @@
 import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
 import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.ResourceOperations;
 import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 import io.javaoperatorsdk.operator.processing.event.source.EventSource;
@@ -79,7 +80,11 @@ public UpdateControl<OnRelistFilterCustomResource> reconcile(
     if (execution == 1) {
       var cm = prepareConfigMap(resource);
       switch (mode.get()) {
-        case NO_RELIST -> context.resourceOperations().serverSideApply(cm, configMapEventSource);
+        case NO_RELIST ->
+            context
+                .resourceOperations()
+                .serverSideApply(
+                    cm, configMapEventSource, ResourceOperations.Options.forcedFiltering());
         case RELIST_AROUND_UPDATE -> {
           configMapEventSource.simulateOnBeforeList();
           var applied = context.resourceOperations().serverSideApply(cm, configMapEventSource);
@@ -94,7 +99,10 @@ public UpdateControl<OnRelistFilterCustomResource> reconcile(
         case RELIST_COMPLETES_BEFORE_UPDATE -> {
           configMapEventSource.simulateOnBeforeList();
           configMapEventSource.simulateOnList();
-          context.resourceOperations().serverSideApply(cm, configMapEventSource);
+          context
+              .resourceOperations()
+              .serverSideApply(
+                  cm, configMapEventSource, ResourceOperations.Options.forcedFiltering());
         }
         case RELIST_STARTS_DURING_UPDATE -> {
           // Drive the event-filtering update path manually so we can fire onBeforeList AFTER the
@@ -114,7 +122,8 @@ public UpdateControl<OnRelistFilterCustomResource> reconcile(
                                 .withFieldManager(fieldManager)
                                 .withPatchType(PatchType.SERVER_SIDE_APPLY)
                                 .build());
-                  });
+                  },
+                  true);
           // See RELIST_AROUND_UPDATE: wait for the own-write event to be buffered while the
           // re-list is still in progress, so it is tagged as part of the re-list and propagated.
           configMapEventSource.awaitWatchEventReceived(applied);

operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/ownsecondaryupdate/OwnSecondaryUpdateReconciler.java

@@ -27,6 +27,7 @@
 import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
 import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.ResourceOperations;
 import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
 import io.javaoperatorsdk.operator.processing.event.source.EventSource;
 import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
@@ -50,7 +51,12 @@ public UpdateControl<OwnSecondaryUpdateCustomResource> reconcile(
     // version actually advances. With the read-cache-after-write filter in place, none of the
     // resulting watch events should trigger a fresh reconciliation.
     for (int i = 1; i <= OWN_SSA_COUNT; i++) {
-      context.resourceOperations().serverSideApply(prepareCM(resource, i), configMapEventSource);
+      context
+          .resourceOperations()
+          .serverSideApply(
+              prepareCM(resource, i),
+              configMapEventSource,
+              ResourceOperations.Options.forcedFiltering());
     }
     return UpdateControl.noUpdate();
   }

operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/specchangeduringstatuspatch/SpecChangeDuringStatusPatchCustomResource.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright Java Operator SDK Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.javaoperatorsdk.operator.baseapi.readcacheafterwrite.specchangeduringstatuspatch;
+
+import io.fabric8.kubernetes.api.model.Namespaced;
+import io.fabric8.kubernetes.client.CustomResource;
+import io.fabric8.kubernetes.model.annotation.Group;
+import io.fabric8.kubernetes.model.annotation.ShortNames;
+import io.fabric8.kubernetes.model.annotation.Version;
+
+@Group("sample.javaoperatorsdk")
+@Version("v1")
+@ShortNames("scdsp")
+public class SpecChangeDuringStatusPatchCustomResource
+    extends CustomResource<SpecChangeDuringStatusPatchSpec, SpecChangeDuringStatusPatchStatus>
+    implements Namespaced {}