feat: add feature switch to use SSA for managing finalizer

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/ReconcilerUtils.java

@@ -6,14 +6,11 @@
 import java.lang.reflect.Method;
 import java.util.Arrays;
 import java.util.Locale;
-import java.util.Objects;
 import java.util.function.Predicate;
 import java.util.regex.Pattern;
 
 import io.fabric8.kubernetes.api.builder.Builder;
-import io.fabric8.kubernetes.api.model.GenericKubernetesResource;
 import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.fabric8.kubernetes.api.model.Namespaced;
 import io.fabric8.kubernetes.client.CustomResource;
 import io.fabric8.kubernetes.client.KubernetesClientException;
 import io.fabric8.kubernetes.client.utils.Serialization;
@@ -73,36 +70,6 @@ public static String getNameFor(Class<? extends Reconciler> reconcilerClass) {
     return getDefaultNameFor(reconcilerClass);
   }
 
-  public static void checkIfCanAddOwnerReference(HasMetadata owner, HasMetadata resource) {
-    if (owner instanceof GenericKubernetesResource
-        || resource instanceof GenericKubernetesResource) {
-      return;
-    }
-    if (owner instanceof Namespaced) {
-      if (!(resource instanceof Namespaced)) {
-        throw new OperatorException(
-            "Cannot add owner reference from a cluster scoped to a namespace scoped resource."
-                + resourcesIdentifierDescription(owner, resource));
-      } else if (!Objects.equals(
-          owner.getMetadata().getNamespace(), resource.getMetadata().getNamespace())) {
-        throw new OperatorException(
-            "Cannot add owner reference between two resource in different namespaces."
-                + resourcesIdentifierDescription(owner, resource));
-      }
-    }
-  }
-
-  private static String resourcesIdentifierDescription(HasMetadata owner, HasMetadata resource) {
-    return " Owner name: "
-        + owner.getMetadata().getName()
-        + " Kind: "
-        + owner.getKind()
-        + ", Resource name: "
-        + resource.getMetadata().getName()
-        + " Kind: "
-        + resource.getKind();
-  }
-
   public static String getNameFor(Reconciler reconciler) {
     return getNameFor(reconciler.getClass());
   }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java

@@ -493,8 +493,7 @@ default boolean parseResourceVersionsForEventFilteringAndCaching() {
 
   /**
    * {@link io.javaoperatorsdk.operator.api.reconciler.UpdateControl} patch resource or status can
-   * either use simple patches or SSA. Setting this to {@code true}, controllers will use SSA for
-   * adding finalizers, patching resources and status.
+   * either use simple patches or SSA. Setting this to {@code true}, patching resources and status.
    *
    * @return {@code true} if Server-Side Apply (SSA) should be used when patching the primary
    *     resources, {@code false} otherwise
@@ -505,6 +504,18 @@ default boolean useSSAToPatchPrimaryResource() {
     return true;
   }
 
+  /**
+   * Setting this to {@code true}, controllers will use SSA for adding finalizers.
+   *
+   * @return {@code true} if Server-Side Apply (SSA) should be used when managing finalizers, {@code
+   *     false} otherwise
+   * @see ConfigurationServiceOverrider#withUseSSAToAddFinalizer(boolean)
+   * @since 5.1.2
+   */
+  default boolean useSSAToAddFinalizer() {
+    return true;
+  }
+
   /**
    * Determines whether resources retrieved from caches such as via calls to {@link
    * Context#getSecondaryResource(Class)} should be defensively cloned first.

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java

@@ -39,6 +39,7 @@ public class ConfigurationServiceOverrider {
   private Boolean previousAnnotationForDependentResources;
   private Boolean parseResourceVersions;
   private Boolean useSSAToPatchPrimaryResource;
+  private Boolean useSSAToAddFinalizer;
   private Boolean cloneSecondaryResourcesWhenGettingFromCache;
   private Set<Class<? extends HasMetadata>> previousAnnotationUsageBlocklist;
 
@@ -183,6 +184,11 @@ public ConfigurationServiceOverrider withUseSSAToPatchPrimaryResource(boolean va
     return this;
   }
 
+  public ConfigurationServiceOverrider withUseSSAToAddFinalizer(boolean value) {
+    this.useSSAToAddFinalizer = value;
+    return this;
+  }
+
   public ConfigurationServiceOverrider withCloneSecondaryResourcesWhenGettingFromCache(
       boolean value) {
     this.cloneSecondaryResourcesWhenGettingFromCache = value;
@@ -336,6 +342,12 @@ public boolean useSSAToPatchPrimaryResource() {
             useSSAToPatchPrimaryResource, ConfigurationService::useSSAToPatchPrimaryResource);
       }
 
+      @Override
+      public boolean useSSAToAddFinalizer() {
+        return overriddenValueOrDefault(
+            useSSAToAddFinalizer, ConfigurationService::useSSAToPatchPrimaryResource);
+      }
+
       @Override
       public boolean cloneSecondaryResourcesWhenGettingFromCache() {
         return overriddenValueOrDefault(

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/Field.java

@@ -0,0 +1,10 @@
+package io.javaoperatorsdk.operator.api.config.informer;
+
+public @interface Field {
+
+  String path();
+
+  String value();
+
+  boolean negated() default false;
+}

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/FieldSelector.java

@@ -0,0 +1,26 @@
+package io.javaoperatorsdk.operator.api.config.informer;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class FieldSelector {
+  private final List<Field> fields;
+
+  public FieldSelector(List<Field> fields) {
+    this.fields = fields;
+  }
+
+  public FieldSelector(Field... fields) {
+    this.fields = Arrays.asList(fields);
+  }
+
+  public List<Field> getFields() {
+    return fields;
+  }
+
+  public record Field(String path, String value, boolean negated) {
+    public Field(String path, String value) {
+      this(path, value, false);
+    }
+  }
+}

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/FieldSelectorBuilder.java

@@ -0,0 +1,23 @@
+package io.javaoperatorsdk.operator.api.config.informer;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class FieldSelectorBuilder {
+
+  private final List<FieldSelector.Field> fields = new ArrayList<>();
+
+  public FieldSelectorBuilder withField(String path, String value) {
+    fields.add(new FieldSelector.Field(path, value));
+    return this;
+  }
+
+  public FieldSelectorBuilder withoutField(String path, String value) {
+    fields.add(new FieldSelector.Field(path, value, true));
+    return this;
+  }
+
+  public FieldSelector build() {
+    return new FieldSelector(fields);
+  }
+}

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/Informer.java

@@ -113,4 +113,7 @@
    * the informer cache.
    */
   long informerListLimit() default NO_LONG_VALUE_SET;
+
+  /** Kubernetes field selector for additional resource filtering */
+  Field[] fieldSelector() default {};
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerConfiguration.java

@@ -1,5 +1,6 @@
 package io.javaoperatorsdk.operator.api.config.informer;
 
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Set;
@@ -36,6 +37,7 @@ public class InformerConfiguration<R extends HasMetadata> {
   private GenericFilter<? super R> genericFilter;
   private ItemStore<R> itemStore;
   private Long informerListLimit;
+  private FieldSelector fieldSelector;
 
   protected InformerConfiguration(
       Class<R> resourceClass,
@@ -48,7 +50,8 @@ protected InformerConfiguration(
       OnDeleteFilter<? super R> onDeleteFilter,
       GenericFilter<? super R> genericFilter,
       ItemStore<R> itemStore,
-      Long informerListLimit) {
+      Long informerListLimit,
+      FieldSelector fieldSelector) {
     this(resourceClass);
     this.name = name;
     this.namespaces = namespaces;
@@ -60,6 +63,7 @@ protected InformerConfiguration(
     this.genericFilter = genericFilter;
     this.itemStore = itemStore;
     this.informerListLimit = informerListLimit;
+    this.fieldSelector = fieldSelector;
   }
 
   private InformerConfiguration(Class<R> resourceClass) {
@@ -93,7 +97,8 @@ public static <R extends HasMetadata> InformerConfiguration<R>.Builder builder(
             original.onDeleteFilter,
             original.genericFilter,
             original.itemStore,
-            original.informerListLimit)
+            original.informerListLimit,
+            original.fieldSelector)
         .builder;
   }
 
@@ -264,6 +269,10 @@ public Long getInformerListLimit() {
     return informerListLimit;
   }
 
+  public FieldSelector getFieldSelector() {
+    return fieldSelector;
+  }
+
   @SuppressWarnings("UnusedReturnValue")
   public class Builder {
 
@@ -329,6 +338,12 @@ public InformerConfiguration<R>.Builder initFromAnnotation(
         final var informerListLimit =
             informerListLimitValue == Constants.NO_LONG_VALUE_SET ? null : informerListLimitValue;
         withInformerListLimit(informerListLimit);
+
+        withFieldSelector(
+            new FieldSelector(
+                Arrays.stream(informerConfig.fieldSelector())
+                    .map(f -> new FieldSelector.Field(f.path(), f.value(), f.negated()))
+                    .toList()));
       }
       return this;
     }
@@ -424,5 +439,10 @@ public Builder withInformerListLimit(Long informerListLimit) {
       InformerConfiguration.this.informerListLimit = informerListLimit;
       return this;
     }
+
+    public Builder withFieldSelector(FieldSelector fieldSelector) {
+      InformerConfiguration.this.fieldSelector = fieldSelector;
+      return this;
+    }
   }
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerEventSourceConfiguration.java

@@ -265,6 +265,11 @@ public Builder<R> withInformerListLimit(Long informerListLimit) {
       return this;
     }
 
+    public Builder<R> withFieldSelector(FieldSelector fieldSelector) {
+      config.withFieldSelector(fieldSelector);
+      return this;
+    }
+
     public void updateFrom(InformerConfiguration<R> informerConfig) {
       if (informerConfig != null) {
         final var informerConfigName = informerConfig.getName();
@@ -281,7 +286,8 @@ public void updateFrom(InformerConfiguration<R> informerConfig) {
             .withOnUpdateFilter(informerConfig.getOnUpdateFilter())
             .withOnDeleteFilter(informerConfig.getOnDeleteFilter())
             .withGenericFilter(informerConfig.getGenericFilter())
-            .withInformerListLimit(informerConfig.getInformerListLimit());
+            .withInformerListLimit(informerConfig.getInformerListLimit())
+            .withFieldSelector(informerConfig.getFieldSelector());
       }
     }
 

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

@@ -11,7 +11,6 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.api.model.Namespaced;
 import io.fabric8.kubernetes.client.dsl.Resource;
-import io.javaoperatorsdk.operator.ReconcilerUtils;
 import io.javaoperatorsdk.operator.api.config.dependent.Configured;
 import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
@@ -206,7 +205,6 @@ protected Resource<R> prepare(Context<P> context, R desired, P primary, String a
 
   protected void addReferenceHandlingMetadata(R desired, P primary) {
     if (addOwnerReference()) {
-      ReconcilerUtils.checkIfCanAddOwnerReference(primary, desired);
       desired.addOwnerReference(primary);
     } else if (useNonOwnerRefBasedSecondaryToPrimaryMapping()) {
       addSecondaryToPrimaryMapperAnnotations(desired, primary);

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

@@ -42,6 +42,7 @@ class ReconciliationDispatcher<P extends HasMetadata> {
   private final boolean retryConfigurationHasZeroAttempts;
   private final Cloner cloner;
   private final boolean useSSA;
+  private final boolean useSSAToAddFinalizer;
 
   ReconciliationDispatcher(Controller<P> controller, CustomResourceFacade<P> customResourceFacade) {
     this.controller = controller;
@@ -52,6 +53,7 @@ class ReconciliationDispatcher<P extends HasMetadata> {
     var retry = configuration.getRetry();
     retryConfigurationHasZeroAttempts = retry == null || retry.initExecution().isLastAttempt();
     useSSA = configuration.getConfigurationService().useSSAToPatchPrimaryResource();
+    useSSAToAddFinalizer = configuration.getConfigurationService().useSSAToAddFinalizer();
   }
 
   public ReconciliationDispatcher(Controller<P> controller) {
@@ -119,7 +121,7 @@ private PostExecutionControl<P> handleReconcile(
        * finalizer.
        */
       P updatedResource;
-      if (useSSA) {
+      if (useSSAToAddFinalizer) {
         updatedResource = addFinalizerWithSSA(originalResource);
       } else {
         updatedResource = updateCustomResourceWithFinalizer(resourceForExecution, originalResource);

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

@@ -134,6 +134,18 @@ private InformerWrapper<R> createEventSource(
       ResourceEventHandler<R> eventHandler,
       String namespaceIdentifier) {
     final var informerConfig = configuration.getInformerConfig();
+
+    if (informerConfig.getFieldSelector() != null
+        && !informerConfig.getFieldSelector().getFields().isEmpty()) {
+      for (var f : informerConfig.getFieldSelector().getFields()) {
+        if (f.negated()) {
+          filteredBySelectorClient = filteredBySelectorClient.withoutField(f.path(), f.value());
+        } else {
+          filteredBySelectorClient = filteredBySelectorClient.withField(f.path(), f.value());
+        }
+      }
+    }
+
     var informer =
         Optional.ofNullable(informerConfig.getInformerListLimit())
             .map(filteredBySelectorClient::withLimit)

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/ReconcilerUtilsTest.java

@@ -8,8 +8,6 @@
 import io.fabric8.kubernetes.api.model.apps.Deployment;
 import io.fabric8.kubernetes.api.model.apps.DeploymentBuilder;
 import io.fabric8.kubernetes.api.model.apps.DeploymentSpec;
-import io.fabric8.kubernetes.api.model.rbac.ClusterRole;
-import io.fabric8.kubernetes.api.model.rbac.ClusterRoleBuilder;
 import io.fabric8.kubernetes.client.CustomResource;
 import io.fabric8.kubernetes.client.KubernetesClientException;
 import io.fabric8.kubernetes.client.http.HttpRequest;
@@ -154,44 +152,6 @@ void handleKubernetesExceptionShouldThrowMissingCRDExceptionWhenAppropriate() {
                 HasMetadata.getFullResourceName(Tomcat.class)));
   }
 
-  @Test
-  void checksIfOwnerReferenceCanBeAdded() {
-    assertThrows(
-        OperatorException.class,
-        () ->
-            ReconcilerUtils.checkIfCanAddOwnerReference(
-                namespacedResource(), namespacedResourceFromOtherNamespace()));
-
-    assertThrows(
-        OperatorException.class,
-        () ->
-            ReconcilerUtils.checkIfCanAddOwnerReference(
-                namespacedResource(), clusterScopedResource()));
-
-    assertDoesNotThrow(
-        () -> {
-          ReconcilerUtils.checkIfCanAddOwnerReference(
-              clusterScopedResource(), clusterScopedResource());
-          ReconcilerUtils.checkIfCanAddOwnerReference(namespacedResource(), namespacedResource());
-        });
-  }
-
-  private ClusterRole clusterScopedResource() {
-    return new ClusterRoleBuilder().withMetadata(new ObjectMetaBuilder().build()).build();
-  }
-
-  private ConfigMap namespacedResource() {
-    return new ConfigMapBuilder()
-        .withMetadata(new ObjectMetaBuilder().withNamespace("testns1").build())
-        .build();
-  }
-
-  private ConfigMap namespacedResourceFromOtherNamespace() {
-    return new ConfigMapBuilder()
-        .withMetadata(new ObjectMetaBuilder().withNamespace("testns2").build())
-        .build();
-  }
-
   @Group("tomcatoperator.io")
   @Version("v1")
   @ShortNames("tc")