refactor: isolate index handling to BulkDependentResource interface

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

@@ -17,9 +17,8 @@
 import io.javaoperatorsdk.operator.OperatorException;
 import io.javaoperatorsdk.operator.ReconcilerUtils;
 import io.javaoperatorsdk.operator.api.config.dependent.DependentResourceSpec;
-import io.javaoperatorsdk.operator.api.reconciler.Constants;
+import io.javaoperatorsdk.operator.api.reconciler.*;
 import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
-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;
@@ -54,9 +53,8 @@ public AnnotationControllerConfiguration(Reconciler<P> reconciler) {
     this.reconciler = reconciler;
     this.annotation = reconciler.getClass().getAnnotation(ControllerConfiguration.class);
     if (annotation == null) {
-      throw new OperatorException(
-          "Missing mandatory @" + ControllerConfiguration.class.getSimpleName() +
-              " annotation for reconciler:  " + reconciler);
+      throw new OperatorException("Missing mandatory @" + CONTROLLER_CONFIG_ANNOTATION +
+          " annotation for reconciler:  " + reconciler);
     }
   }
 
@@ -247,9 +245,9 @@ public List<DependentResourceSpec> getDependentResources() {
         final var context = "DependentResource of type '" + dependentType.getName() + "'";
         spec = new DependentResourceSpec(dependentType, config, name,
             Set.of(dependent.dependsOn()),
-            instantiateConditionIfNotDefault(dependent.readyPostcondition(), context),
-            instantiateConditionIfNotDefault(dependent.reconcilePrecondition(), context),
-            instantiateConditionIfNotDefault(dependent.deletePostcondition(), context));
+            instantiateIfNotDefault(dependent.readyPostcondition(), Condition.class, context),
+            instantiateIfNotDefault(dependent.reconcilePrecondition(), Condition.class, context),
+            instantiateIfNotDefault(dependent.deletePostcondition(), Condition.class, context));
         specsMap.put(name, spec);
       }
 
@@ -258,10 +256,10 @@ public List<DependentResourceSpec> getDependentResources() {
     return specs;
   }
 
-  protected Condition<?, ?> instantiateConditionIfNotDefault(Class<? extends Condition> condition,
+  protected <T> T instantiateIfNotDefault(Class<? extends T> toInstantiate, Class<T> defaultClass,
       String context) {
-    if (condition != Condition.class) {
-      return instantiateAndConfigureIfNeeded(condition, Condition.class, context);
+    if (!defaultClass.equals(toInstantiate)) {
+      return instantiateAndConfigureIfNeeded(toInstantiate, defaultClass, context);
     }
     return null;
   }
@@ -287,6 +285,7 @@ private Object createKubernetesResourceConfig(Class<? extends DependentResource>
     OnUpdateFilter<? extends HasMetadata> onUpdateFilter = null;
     OnDeleteFilter<? extends HasMetadata> onDeleteFilter = null;
     GenericFilter<? extends HasMetadata> genericFilter = null;
+    ResourceDiscriminator<?, ? extends HasMetadata> resourceDiscriminator = null;
     if (kubeDependent != null) {
       if (!Arrays.equals(KubernetesDependent.DEFAULT_NAMESPACES,
           kubeDependent.namespaces())) {
@@ -297,7 +296,6 @@ private Object createKubernetesResourceConfig(Class<? extends DependentResource>
       final var fromAnnotation = kubeDependent.labelSelector();
       labelSelector = Constants.NO_VALUE_SET.equals(fromAnnotation) ? null : fromAnnotation;
 
-
       final var context =
           KUBE_DEPENDENT_NAME + " annotation on " + dependentType.getName() + " DependentResource";
       onAddFilter = createFilter(kubeDependent.onAddFilter(), OnAddFilter.class, context)
@@ -311,10 +309,15 @@ private Object createKubernetesResourceConfig(Class<? extends DependentResource>
       genericFilter =
           createFilter(kubeDependent.genericFilter(), GenericFilter.class, context)
               .orElse(null);
+
+      resourceDiscriminator =
+          instantiateIfNotDefault(kubeDependent.resourceDiscriminator(),
+              ResourceDiscriminator.class, context);
     }
 
     config =
-        new KubernetesDependentResourceConfig(namespaces, labelSelector, configuredNS, onAddFilter,
+        new KubernetesDependentResourceConfig(namespaces, labelSelector, configuredNS,
+            resourceDiscriminator, onAddFilter,
             onUpdateFilter, onDeleteFilter, genericFilter);
 
     return config;

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/Context.java

@@ -6,20 +6,27 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.ManagedDependentResourceContext;
+import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
 
 public interface Context<P extends HasMetadata> {
 
   Optional<RetryInfo> getRetryInfo();
 
-  default <T> Optional<T> getSecondaryResource(Class<T> expectedType) {
-    return getSecondaryResource(expectedType, null);
+  default <R> Optional<R> getSecondaryResource(Class<R> expectedType) {
+    return getSecondaryResource(expectedType, (String) null);
   }
 
-  <T> Set<T> getSecondaryResources(Class<T> expectedType);
+  <R> Set<R> getSecondaryResources(Class<R> expectedType);
 
-  <T> Optional<T> getSecondaryResource(Class<T> expectedType, String eventSourceName);
+  @Deprecated(forRemoval = true)
+  <R> Optional<R> getSecondaryResource(Class<R> expectedType, String eventSourceName);
+
+  <R> Optional<R> getSecondaryResource(Class<R> expectedType,
+      ResourceDiscriminator<R, P> discriminator);
 
   ControllerConfiguration<P> getControllerConfiguration();
 
   ManagedDependentResourceContext managedDependentResourceContext();
+
+  EventSourceRetriever<P> eventSourceRetriever();
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContext.java

@@ -9,6 +9,7 @@
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DefaultManagedDependentResourceContext;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.ManagedDependentResourceContext;
 import io.javaoperatorsdk.operator.processing.Controller;
+import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
 
 public class DefaultContext<P extends HasMetadata> implements Context<P> {
 
@@ -47,6 +48,12 @@ public <T> Optional<T> getSecondaryResource(Class<T> expectedType, String eventS
         .getSecondaryResource(primaryResource);
   }
 
+  @Override
+  public <R> Optional<R> getSecondaryResource(Class<R> expectedType,
+      ResourceDiscriminator<R, P> discriminator) {
+    return discriminator.distinguish(expectedType, primaryResource, this);
+  }
+
   @Override
   public ControllerConfiguration<P> getControllerConfiguration() {
     return controllerConfiguration;
@@ -57,6 +64,11 @@ public ManagedDependentResourceContext managedDependentResourceContext() {
     return defaultManagedDependentResourceContext;
   }
 
+  @Override
+  public EventSourceRetriever<P> eventSourceRetriever() {
+    return controller.getEventSourceManager();
+  }
+
   public DefaultContext<P> setRetryInfo(RetryInfo retryInfo) {
     this.retryInfo = retryInfo;
     return this;

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ResourceDiscriminator.java

@@ -0,0 +1,10 @@
+package io.javaoperatorsdk.operator.api.reconciler;
+
+import java.util.Optional;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+
+public interface ResourceDiscriminator<R, P extends HasMetadata> {
+
+  Optional<R> distinguish(Class<R> resource, P primary, Context<P> context);
+}

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ResourceIDMatcherDiscriminator.java

@@ -0,0 +1,25 @@
+package io.javaoperatorsdk.operator.api.reconciler;
+
+import java.util.Optional;
+import java.util.function.Function;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+
+public class ResourceIDMatcherDiscriminator<R extends HasMetadata, P extends HasMetadata>
+    implements ResourceDiscriminator<R, P> {
+
+  private final Function<P, ResourceID> mapper;
+
+  public ResourceIDMatcherDiscriminator(Function<P, ResourceID> mapper) {
+    this.mapper = mapper;
+  }
+
+  @Override
+  public Optional<R> distinguish(Class<R> resource, P primary, Context<P> context) {
+    var resourceID = mapper.apply(primary);
+    return context.getSecondaryResources(resource).stream()
+        .filter(resourceID::isSameResource)
+        .findFirst();
+  }
+}

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/DependentResource.java

@@ -1,16 +1,17 @@
 package io.javaoperatorsdk.operator.api.reconciler.dependent;
 
+import java.util.Optional;
+
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
-import io.javaoperatorsdk.operator.processing.ResourceOwner;
 
 /**
  * An interface to implement and provide dependent resource support.
  *
  * @param <R> the dependent resource type
  * @param <P> the associated primary resource type
  */
-public interface DependentResource<R, P extends HasMetadata> extends ResourceOwner<R, P> {
+public interface DependentResource<R, P extends HasMetadata> {
 
   /**
    * Reconciles the dependent resource given the desired primary state
@@ -21,6 +22,17 @@ public interface DependentResource<R, P extends HasMetadata> extends ResourceOwn
    */
   ReconcileResult<R> reconcile(P primary, Context<P> context);
 
+  /**
+   * Retrieves the resource type associated with this DependentResource
+   *
+   * @return the resource type associated with this DependentResource
+   */
+  Class<R> resourceType();
+
+  default Optional<R> getSecondaryResource(P primary, Context<P> context) {
+    return Optional.empty();
+  }
+
   /**
    * Computes a default name for the specified DependentResource class
    *

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/ReconcileResult.java

@@ -1,14 +1,14 @@
 package io.javaoperatorsdk.operator.api.reconciler.dependent;
 
-import java.util.Optional;
+import java.util.*;
+import java.util.stream.Collectors;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 
 public class ReconcileResult<R> {
 
-  private final R resource;
-  private final Operation operation;
+  private final Map<R, Operation> resourceOperations;
 
   public static <T> ReconcileResult<T> resourceCreated(T resource) {
     return new ReconcileResult<>(resource, Operation.CREATED);
@@ -22,25 +22,49 @@ public static <T> ReconcileResult<T> noOperation(T resource) {
     return new ReconcileResult<>(resource, Operation.NONE);
   }
 
+  @SafeVarargs
+  public static <T> ReconcileResult<T> aggregatedResult(ReconcileResult<T>... results) {
+    if (results == null) {
+      throw new IllegalArgumentException("Should provide results to aggregate");
+    }
+    if (results.length == 1) {
+      return results[0];
+    }
+    final Map<T, Operation> operations = new HashMap<>(results.length);
+    for (ReconcileResult<T> res : results) {
+      res.getSingleResource().ifPresent(r -> operations.put(r, res.getSingleOperation()));
+    }
+    return new ReconcileResult<>(operations);
+  }
+
   @Override
   public String toString() {
-    return getResource()
-        .map(r -> r instanceof HasMetadata ? ResourceID.fromResource((HasMetadata) r) : r)
-        .orElse("no resource")
-        + " -> " + operation;
+    return resourceOperations.entrySet().stream().collect(Collectors.toMap(
+        e -> e instanceof HasMetadata ? ResourceID.fromResource((HasMetadata) e) : e,
+        Map.Entry::getValue))
+        .toString();
   }
 
   private ReconcileResult(R resource, Operation operation) {
-    this.resource = resource;
-    this.operation = operation;
+    resourceOperations = resource != null ? Map.of(resource, operation) : Collections.emptyMap();
+  }
+
+  private ReconcileResult(Map<R, Operation> operations) {
+    resourceOperations = Collections.unmodifiableMap(operations);
+  }
+
+  public Optional<R> getSingleResource() {
+    return resourceOperations.entrySet().stream().findFirst().map(Map.Entry::getKey);
   }
 
-  public Optional<R> getResource() {
-    return Optional.ofNullable(resource);
+  public Operation getSingleOperation() {
+    return resourceOperations.entrySet().stream().findFirst().map(Map.Entry::getValue)
+        .orElseThrow();
   }
 
-  public Operation getOperation() {
-    return operation;
+  @SuppressWarnings("unused")
+  public Map<R, Operation> getResourceOperations() {
+    return resourceOperations;
   }
 
   public enum Operation {