feat: introduce AnnotationDependentResourceConfigurator concept

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

@@ -2,7 +2,6 @@
 
 import java.lang.annotation.Annotation;
 import java.time.Duration;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -20,16 +19,13 @@
 import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.Dependent;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
-import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependent;
-import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResource;
-import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResourceConfig;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.AnnotationDependentResourceConfigurator;
 import io.javaoperatorsdk.operator.processing.dependent.workflow.Condition;
 import io.javaoperatorsdk.operator.processing.event.rate.RateLimiter;
 import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEventFilter;
 import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEventFilters;
 import io.javaoperatorsdk.operator.processing.event.source.filter.GenericFilter;
 import io.javaoperatorsdk.operator.processing.event.source.filter.OnAddFilter;
-import io.javaoperatorsdk.operator.processing.event.source.filter.OnDeleteFilter;
 import io.javaoperatorsdk.operator.processing.event.source.filter.OnUpdateFilter;
 import io.javaoperatorsdk.operator.processing.retry.Retry;
 
@@ -159,7 +155,7 @@ public Retry getRetry() {
 
 
   @SuppressWarnings("unchecked")
-  private <T> void configureFromAnnotatedReconciler(T instance) {
+  private void configureFromAnnotatedReconciler(Object instance) {
     if (instance instanceof AnnotationConfigurable) {
       AnnotationConfigurable configurable = (AnnotationConfigurable) instance;
       final Class<? extends Annotation> configurationClass =
@@ -172,6 +168,22 @@ private <T> void configureFromAnnotatedReconciler(T instance) {
     }
   }
 
+  @SuppressWarnings("unchecked")
+  private void configureFromCustomAnnotation(Object instance) {
+    if (instance instanceof AnnotationDependentResourceConfigurator) {
+      AnnotationDependentResourceConfigurator configurator =
+          (AnnotationDependentResourceConfigurator) instance;
+      final Class<? extends Annotation> configurationClass =
+          (Class<? extends Annotation>) Utils.getFirstTypeArgumentFromInterface(
+              instance.getClass(), AnnotationDependentResourceConfigurator.class);
+      final var configAnnotation = instance.getClass().getAnnotation(configurationClass);
+      // always called even if the annotation is null so that implementations can provide default
+      // values
+      final var config = configurator.configFrom(configAnnotation, this);
+      configurator.configureWith(config);
+    }
+  }
+
   @Override
   @SuppressWarnings("unchecked")
   public Optional<OnAddFilter<P>> onAddFilter() {
@@ -209,24 +221,29 @@ public List<DependentResourceSpec> getDependentResources() {
 
       final var specsMap = new LinkedHashMap<String, DependentResourceSpec>(dependents.length);
       for (Dependent dependent : dependents) {
-        Object config = null;
         final Class<? extends DependentResource> dependentType = dependent.type();
-        if (KubernetesDependentResource.class.isAssignableFrom(dependentType)) {
-          config = createKubernetesResourceConfig(dependentType);
-        }
 
         final var name = getName(dependent, dependentType);
         var spec = specsMap.get(name);
         if (spec != null) {
           throw new IllegalArgumentException(
               "A DependentResource named '" + name + "' already exists: " + spec);
         }
+
+        final var dependentResource = Utils.instantiateAndConfigureIfNeeded(dependentType,
+            DependentResource.class,
+            Utils.contextFor(this, dependentType, Dependent.class),
+            this::configureFromCustomAnnotation);
+
+        var eventSourceName = dependent.useEventSourceWithName();
+        eventSourceName = Constants.NO_VALUE_SET.equals(eventSourceName) ? null : eventSourceName;
         final var context = Utils.contextFor(this, dependentType, null);
-        spec = new DependentResourceSpec(dependentType, config, name,
+        spec = new DependentResourceSpec(dependentResource, name,
             Set.of(dependent.dependsOn()),
             Utils.instantiate(dependent.readyPostcondition(), Condition.class, context),
             Utils.instantiate(dependent.reconcilePrecondition(), Condition.class, context),
-            Utils.instantiate(dependent.deletePostcondition(), Condition.class, context));
+            Utils.instantiate(dependent.deletePostcondition(), Condition.class, context),
+            eventSourceName);
         specsMap.put(name, spec);
       }
 
@@ -243,48 +260,6 @@ private String getName(Dependent dependent, Class<? extends DependentResource> d
     return name;
   }
 
-  @SuppressWarnings({"rawtypes", "unchecked"})
-  private Object createKubernetesResourceConfig(Class<? extends DependentResource> dependentType) {
-
-    Object config;
-    final var kubeDependent = dependentType.getAnnotation(KubernetesDependent.class);
-
-    var namespaces = getNamespaces();
-    var configuredNS = false;
-    String labelSelector = null;
-    OnAddFilter<? extends HasMetadata> onAddFilter = null;
-    OnUpdateFilter<? extends HasMetadata> onUpdateFilter = null;
-    OnDeleteFilter<? extends HasMetadata> onDeleteFilter = null;
-    GenericFilter<? extends HasMetadata> genericFilter = null;
-    if (kubeDependent != null) {
-      if (!Arrays.equals(KubernetesDependent.DEFAULT_NAMESPACES,
-          kubeDependent.namespaces())) {
-        namespaces = Set.of(kubeDependent.namespaces());
-        configuredNS = true;
-      }
-
-      final var fromAnnotation = kubeDependent.labelSelector();
-      labelSelector = Constants.NO_VALUE_SET.equals(fromAnnotation) ? null : fromAnnotation;
-
-
-      final var context =
-          Utils.contextFor(this, dependentType, null);
-      onAddFilter = Utils.instantiate(kubeDependent.onAddFilter(), OnAddFilter.class, context);
-      onUpdateFilter =
-          Utils.instantiate(kubeDependent.onUpdateFilter(), OnUpdateFilter.class, context);
-      onDeleteFilter =
-          Utils.instantiate(kubeDependent.onDeleteFilter(), OnDeleteFilter.class, context);
-      genericFilter =
-          Utils.instantiate(kubeDependent.genericFilter(), GenericFilter.class, context);
-    }
-
-    config =
-        new KubernetesDependentResourceConfig(namespaces, labelSelector, configuredNS, onAddFilter,
-            onUpdateFilter, onDeleteFilter, genericFilter);
-
-    return config;
-  }
-
   public static <T> T valueOrDefault(
       ControllerConfiguration controllerConfiguration,
       Function<ControllerConfiguration, T> mapper,

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

@@ -140,8 +140,9 @@ default ObjectMapper getObjectMapper() {
     return Serialization.jsonMapper();
   }
 
+  @Deprecated(forRemoval = true)
   default DependentResourceFactory dependentResourceFactory() {
-    return new DependentResourceFactory() {};
+    return null;
   }
 
   default Optional<LeaderElectionConfiguration> getLeaderElectionConfiguration() {

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

@@ -6,12 +6,11 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.Set;
-import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.config.dependent.DependentResourceSpec;
-import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResourceConfig;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DependentResourceConfigurator;
 import io.javaoperatorsdk.operator.processing.event.rate.RateLimiter;
 import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEventFilter;
 import io.javaoperatorsdk.operator.processing.event.source.filter.GenericFilter;
@@ -159,42 +158,39 @@ public ControllerConfigurationOverrider<R> withGenericFilter(GenericFilter<R> ge
     return this;
   }
 
+  @SuppressWarnings("unchecked")
   public ControllerConfigurationOverrider<R> replacingNamedDependentResourceConfig(String name,
       Object dependentResourceConfig) {
 
     var current = namedDependentResourceSpecs.get(name);
     if (current == null) {
       throw new IllegalArgumentException("Cannot find a DependentResource named: " + name);
     }
-    replaceConfig(name, dependentResourceConfig, current);
-    return this;
-  }
 
-  private void replaceConfig(String name, Object newConfig, DependentResourceSpec<?, ?> current) {
-    namedDependentResourceSpecs.put(name,
-        new DependentResourceSpec<>(current.getDependentResourceClass(), newConfig, name,
-            current.getDependsOn(), current.getReadyCondition(), current.getReconcileCondition(),
-            current.getDeletePostCondition()));
+    var dependentResource = current.getDependentResource();
+    if (dependentResource instanceof DependentResourceConfigurator) {
+      var configurator = (DependentResourceConfigurator) dependentResource;
+      configurator.configureWith(dependentResourceConfig);
+    }
+
+    return this;
   }
 
-  @SuppressWarnings("unchecked")
   public ControllerConfiguration<R> build() {
-    // propagate namespaces if needed
-    final List<DependentResourceSpec> newDependentSpecs;
     final var hasModifiedNamespaces = !original.getNamespaces().equals(namespaces);
-    newDependentSpecs = namedDependentResourceSpecs.entrySet().stream()
-        .map(drsEntry -> {
-          final var spec = drsEntry.getValue();
-
-          // if the spec has a config and it's a KubernetesDependentResourceConfig, update the
-          // namespaces if needed, otherwise, just return the existing spec
-          final Optional<?> maybeConfig = spec.getDependentResourceConfiguration();
-          return maybeConfig.filter(KubernetesDependentResourceConfig.class::isInstance)
-              .map(KubernetesDependentResourceConfig.class::cast)
-              .filter(Predicate.not(KubernetesDependentResourceConfig::wereNamespacesConfigured))
-              .map(c -> updateSpec(drsEntry.getKey(), spec, c))
-              .orElse(drsEntry.getValue());
-        }).collect(Collectors.toUnmodifiableList());
+    final var newDependentSpecs = namedDependentResourceSpecs.values().stream()
+        .peek(spec -> {
+          // if the dependent resource has a NamespaceChangeable config
+          // update the namespaces if needed, otherwise, do nothing
+          if (hasModifiedNamespaces) {
+            final Optional<?> maybeConfig = spec.getDependentResourceConfiguration();
+            maybeConfig
+                .filter(NamespaceChangeable.class::isInstance)
+                .map(NamespaceChangeable.class::cast)
+                .filter(NamespaceChangeable::allowsNamespaceChanges)
+                .ifPresent(nc -> nc.changeNamespaces(namespaces));
+          }
+        }).collect(Collectors.toList());
 
     return new DefaultControllerConfiguration<>(
         original.getAssociatedReconcilerClassName(),
@@ -215,14 +211,6 @@ public ControllerConfiguration<R> build() {
         newDependentSpecs);
   }
 
-  @SuppressWarnings({"rawtypes", "unchecked"})
-  private DependentResourceSpec<?, ?> updateSpec(String name, DependentResourceSpec spec,
-      KubernetesDependentResourceConfig c) {
-    return new DependentResourceSpec(spec.getDependentResourceClass(),
-        c.setNamespaces(namespaces), name, spec.getDependsOn(), spec.getReadyCondition(),
-        spec.getReconcileCondition(), spec.getDeletePostCondition());
-  }
-
   public static <R extends HasMetadata> ControllerConfigurationOverrider<R> override(
       ControllerConfiguration<R> original) {
     return new ControllerConfigurationOverrider<>(original);

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

@@ -16,9 +16,9 @@ public interface NamespaceChangeable {
    */
   void changeNamespaces(Set<String> namespaces);
 
+  @SuppressWarnings("unused")
   default void changeNamespaces(String... namespaces) {
-    changeNamespaces(
-        namespaces != null ? Set.of(namespaces) : DEFAULT_NAMESPACES_SET);
+    changeNamespaces(namespaces != null ? Set.of(namespaces) : DEFAULT_NAMESPACES_SET);
   }
 
   default boolean allowsNamespaceChanges() {

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

@@ -111,6 +111,11 @@ public static Class<?> getFirstTypeArgumentFromExtendedClass(Class<?> clazz) {
 
   public static Class<?> getFirstTypeArgumentFromInterface(Class<?> clazz,
       Class<?> expectedImplementedInterface) {
+    return getTypeArgumentFromInterfaceByIndex(clazz, expectedImplementedInterface, 0);
+  }
+
+  public static Class<?> getTypeArgumentFromInterfaceByIndex(Class<?> clazz,
+      Class<?> expectedImplementedInterface, int index) {
     if (expectedImplementedInterface.isAssignableFrom(clazz)) {
       final var genericInterfaces = clazz.getGenericInterfaces();
       Optional<? extends Class<?>> target = Optional.empty();
@@ -122,7 +127,7 @@ public static Class<?> getFirstTypeArgumentFromInterface(Class<?> clazz,
             .map(ParameterizedType.class::cast)
             .findFirst()
             .map(t -> {
-              final Type argument = t.getActualTypeArguments()[0];
+              final Type argument = t.getActualTypeArguments()[index];
               if (argument instanceof Class) {
                 return (Class<?>) argument;
               }
@@ -148,7 +153,7 @@ public static Class<?> getFirstTypeArgumentFromInterface(Class<?> clazz,
       // try the parent
       var parent = clazz.getSuperclass();
       if (!Object.class.equals(parent)) {
-        return getFirstTypeArgumentFromInterface(parent, expectedImplementedInterface);
+        return getTypeArgumentFromInterfaceByIndex(parent, expectedImplementedInterface, index);
       }
     }
     throw new IllegalArgumentException("Couldn't retrieve generic parameter type from "

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

@@ -4,14 +4,14 @@
 import java.util.Optional;
 import java.util.Set;
 
+import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DependentResourceConfigurator;
 import io.javaoperatorsdk.operator.processing.dependent.workflow.Condition;
 
-public class DependentResourceSpec<T extends DependentResource<?, ?>, C> {
+public class DependentResourceSpec<R, P extends HasMetadata, C> {
 
-  private final Class<T> dependentResourceClass;
-
-  private final C dependentResourceConfig;
+  private final DependentResource<R, P> dependentResource;
 
   private final String name;
 
@@ -23,24 +23,43 @@ public class DependentResourceSpec<T extends DependentResource<?, ?>, C> {
 
   private final Condition<?, ?> deletePostCondition;
 
-  public DependentResourceSpec(Class<T> dependentResourceClass, C dependentResourceConfig,
+  private final String useEventSourceWithName;
+
+  public DependentResourceSpec(DependentResource<R, P> dependentResource,
       String name, Set<String> dependsOn, Condition<?, ?> readyCondition,
-      Condition<?, ?> reconcileCondition, Condition<?, ?> deletePostCondition) {
-    this.dependentResourceClass = dependentResourceClass;
-    this.dependentResourceConfig = dependentResourceConfig;
+      Condition<?, ?> reconcileCondition, Condition<?, ?> deletePostCondition,
+      String useEventSourceWithName) {
+    this.dependentResource = dependentResource;
     this.name = name;
     this.dependsOn = dependsOn;
     this.readyCondition = readyCondition;
     this.reconcileCondition = reconcileCondition;
     this.deletePostCondition = deletePostCondition;
+    this.useEventSourceWithName = useEventSourceWithName;
   }
 
-  public Class<T> getDependentResourceClass() {
-    return dependentResourceClass;
+  public DependentResourceSpec(DependentResourceSpec<R, P, C> other) {
+    this.dependentResource = other.dependentResource;
+    this.name = other.name;
+    this.dependsOn = other.dependsOn;
+    this.readyCondition = other.readyCondition;
+    this.reconcileCondition = other.reconcileCondition;
+    this.deletePostCondition = other.deletePostCondition;
+    this.useEventSourceWithName = other.useEventSourceWithName;
   }
 
+  @SuppressWarnings("unchecked")
+  public Class<DependentResource<R, P>> getDependentResourceClass() {
+    return (Class<DependentResource<R, P>>) dependentResource.getClass();
+  }
+
+  @SuppressWarnings({"unchecked", "rawtypes"})
   public Optional<C> getDependentResourceConfiguration() {
-    return Optional.ofNullable(dependentResourceConfig);
+    if (dependentResource instanceof DependentResourceConfigurator) {
+      var configurator = (DependentResourceConfigurator) dependentResource;
+      return configurator.configuration();
+    }
+    return Optional.empty();
   }
 
   public String getName() {
@@ -50,8 +69,7 @@ public String getName() {
   @Override
   public String toString() {
     return "DependentResourceSpec{ name='" + name +
-        "', type=" + dependentResourceClass.getCanonicalName() +
-        ", config=" + dependentResourceConfig + '}';
+        "', type=" + getDependentResourceClass().getCanonicalName() + '}';
   }
 
   @Override
@@ -62,7 +80,7 @@ public boolean equals(Object o) {
     if (o == null || getClass() != o.getClass()) {
       return false;
     }
-    DependentResourceSpec<?, ?> that = (DependentResourceSpec<?, ?>) o;
+    DependentResourceSpec<?, ?, ?> that = (DependentResourceSpec<?, ?, ?>) o;
     return name.equals(that.name);
   }
 
@@ -89,4 +107,12 @@ public Condition getReconcileCondition() {
   public Condition getDeletePostCondition() {
     return deletePostCondition;
   }
+
+  public DependentResource<R, P> getDependentResource() {
+    return dependentResource;
+  }
+
+  public Optional<String> getUseEventSourceWithName() {
+    return Optional.ofNullable(useEventSourceWithName);
+  }
 }

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

@@ -164,7 +164,7 @@ public <P extends HasMetadata> InformerConfigurationBuilder<R> withNamespacesInh
     }
 
     /**
-     * Whether or not the associated informer should track changes made to the parent
+     * Whether the associated informer should track changes made to the parent
      * {@link io.javaoperatorsdk.operator.processing.Controller}'s namespaces configuration.
      *
      * @param followChanges {@code true} to reconfigure the associated informer when the parent