diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/AnnotationControllerConfiguration.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/AnnotationControllerConfiguration.java
index b5e3fffcf0..dbd09a32cc 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/AnnotationControllerConfiguration.java
+++ b/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;
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/Context.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/Context.java
index 845810c8a1..2e4fb98e6f 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/Context.java
+++ b/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();
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContext.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContext.java
index afb37a8c53..cb7f4ae63b 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContext.java
+++ b/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;
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ResourceDiscriminator.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ResourceDiscriminator.java
new file mode 100644
index 0000000000..072e7d8078
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ResourceDiscriminator.java
@@ -0,0 +1,11 @@
+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);
+
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ResourceIDMatcherDiscriminator.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ResourceIDMatcherDiscriminator.java
new file mode 100644
index 0000000000..f28633252a
--- /dev/null
+++ b/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();
+  }
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/DependentResource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/DependentResource.java
index 0923d19473..8d31778488 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/DependentResource.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/DependentResource.java
@@ -1,8 +1,9 @@
 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.
@@ -10,7 +11,7 @@
  * @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
    *
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/ResourceOwner.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/ResourceOwner.java
deleted file mode 100644
index f9c02a8a33..0000000000
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/ResourceOwner.java
+++ /dev/null
@@ -1,26 +0,0 @@
-package io.javaoperatorsdk.operator.processing;
-
-import java.util.Optional;
-
-import io.fabric8.kubernetes.api.model.HasMetadata;
-
-public interface ResourceOwner<R, P extends HasMetadata> {
-
-  /**
-   * Retrieves the resource type associated with this ResourceOwner
-   *
-   * @return the resource type associated with this ResourceOwner
-   */
-  Class<R> resourceType();
-
-  /**
-   * Retrieves the resource associated with the specified primary one, returning the actual state of
-   * the resource. Typically, this state might come from a local cache, updated after
-   * reconciliation.
-   *
-   * @param primary the primary resource for which we want to retrieve the secondary resource
-   * @return an {@link Optional} containing the secondary resource or {@link Optional#empty()} if it
-   *         doesn't exist
-   */
-  Optional<R> getSecondaryResource(P primary);
-}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractDependentResource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractDependentResource.java
index 5dbdba9358..1abfb3df4b 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractDependentResource.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractDependentResource.java
@@ -1,11 +1,14 @@
 package io.javaoperatorsdk.operator.processing.dependent;
 
+import java.util.Optional;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.Ignore;
+import io.javaoperatorsdk.operator.api.reconciler.ResourceDiscriminator;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.ReconcileResult;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
@@ -21,6 +24,8 @@ public abstract class AbstractDependentResource<R, P extends HasMetadata>
   protected Creator<R, P> creator;
   protected Updater<R, P> updater;
 
+  private ResourceDiscriminator<R, P> resourceDiscriminator;
+
   @SuppressWarnings("unchecked")
   public AbstractDependentResource() {
     creator = creatable ? (Creator<R, P>) this : null;
@@ -29,7 +34,7 @@ public AbstractDependentResource() {
 
   @Override
   public ReconcileResult<R> reconcile(P primary, Context<P> context) {
-    var maybeActual = getSecondaryResource(primary);
+    Optional<R> maybeActual = getSecondaryResource(primary, context);
     if (creatable || updatable) {
       if (maybeActual.isEmpty()) {
         if (creatable) {
@@ -62,6 +67,11 @@ public ReconcileResult<R> reconcile(P primary, Context<P> context) {
     return ReconcileResult.noOperation(maybeActual.orElse(null));
   }
 
+  public Optional<R> getSecondaryResource(P primary, Context<P> context) {
+    return resourceDiscriminator == null ? context.getSecondaryResource(resourceType())
+        : resourceDiscriminator.distinguish(resourceType(), primary, context);
+  }
+
   private void throwIfNull(R desired, P primary, String descriptor) {
     if (desired == null) {
       throw new DependentResourceException(
@@ -118,4 +128,13 @@ protected R desired(P primary, Context<P> context) {
     throw new IllegalStateException(
         "desired method must be implemented if this DependentResource can be created and/or updated");
   }
+
+  public void setResourceDiscriminator(
+      ResourceDiscriminator<R, P> resourceDiscriminator) {
+    this.resourceDiscriminator = resourceDiscriminator;
+  }
+
+  public ResourceDiscriminator<R, P> getResourceDiscriminator() {
+    return resourceDiscriminator;
+  }
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractEventSourceHolderDependentResource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractEventSourceHolderDependentResource.java
index 0ceba16826..be0db98393 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractEventSourceHolderDependentResource.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractEventSourceHolderDependentResource.java
@@ -19,12 +19,16 @@ public abstract class AbstractEventSourceHolderDependentResource<R, P extends Ha
     implements EventSourceProvider<P> {
 
   private T eventSource;
+  private final Class<R> resourceType;
   private boolean isCacheFillerEventSource;
   protected OnAddFilter<R> onAddFilter;
   protected OnUpdateFilter<R> onUpdateFilter;
   protected OnDeleteFilter<R> onDeleteFilter;
   protected GenericFilter<R> genericFilter;
 
+  protected AbstractEventSourceHolderDependentResource(Class<R> resourceType) {
+    this.resourceType = resourceType;
+  }
 
   public EventSource initEventSource(EventSourceContext<P> context) {
     // some sub-classes (e.g. KubernetesDependentResource) can have their event source created
@@ -42,6 +46,11 @@ public EventSource initEventSource(EventSourceContext<P> context) {
     return eventSource;
   }
 
+  @Override
+  public Class<R> resourceType() {
+    return resourceType;
+  }
+
   protected abstract T createEventSource(EventSourceContext<P> context);
 
   protected void setEventSource(T eventSource) {
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractCachingDependentResource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractCachingDependentResource.java
deleted file mode 100644
index 242625bc5d..0000000000
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractCachingDependentResource.java
+++ /dev/null
@@ -1,29 +0,0 @@
-package io.javaoperatorsdk.operator.processing.dependent.external;
-
-import java.util.Optional;
-
-import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.javaoperatorsdk.operator.api.reconciler.Ignore;
-import io.javaoperatorsdk.operator.processing.dependent.AbstractEventSourceHolderDependentResource;
-import io.javaoperatorsdk.operator.processing.event.source.ExternalResourceCachingEventSource;
-
-@Ignore
-public abstract class AbstractCachingDependentResource<R, P extends HasMetadata>
-    extends
-    AbstractEventSourceHolderDependentResource<R, P, ExternalResourceCachingEventSource<R, P>> {
-  private final Class<R> resourceType;
-
-  protected AbstractCachingDependentResource(Class<R> resourceType) {
-    this.resourceType = resourceType;
-  }
-
-  @Override
-  public Class<R> resourceType() {
-    return resourceType;
-  }
-
-  @Override
-  public Optional<R> getSecondaryResource(P primaryResource) {
-    return eventSource().getSecondaryResource(primaryResource);
-  }
-}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractPollingDependentResource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractPollingDependentResource.java
index 6bab6f48cf..2ccba025a7 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractPollingDependentResource.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractPollingDependentResource.java
@@ -2,11 +2,15 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.Ignore;
+import io.javaoperatorsdk.operator.processing.dependent.AbstractEventSourceHolderDependentResource;
 import io.javaoperatorsdk.operator.processing.event.source.CacheKeyMapper;
+import io.javaoperatorsdk.operator.processing.event.source.ExternalResourceCachingEventSource;
 
 @Ignore
 public abstract class AbstractPollingDependentResource<R, P extends HasMetadata>
-    extends AbstractCachingDependentResource<R, P> implements CacheKeyMapper<R> {
+    extends
+    AbstractEventSourceHolderDependentResource<R, P, ExternalResourceCachingEventSource<R, P>>
+    implements CacheKeyMapper<R> {
 
   public static final int DEFAULT_POLLING_PERIOD = 5000;
   private long pollingPeriod;
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractSimpleDependentResource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractSimpleDependentResource.java
index 0675db5f1a..748452c30c 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractSimpleDependentResource.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractSimpleDependentResource.java
@@ -33,7 +33,7 @@ public AbstractSimpleDependentResource(UpdatableCache<R> cache) {
   }
 
   @Override
-  public Optional<R> getSecondaryResource(HasMetadata primaryResource) {
+  public Optional<R> getSecondaryResource(P primaryResource, Context<P> context) {
     return cache.get(ResourceID.fromResource(primaryResource));
   }
 
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependent.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependent.java
index f66ff95373..603f4ae62e 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependent.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependent.java
@@ -6,6 +6,8 @@
 import java.lang.annotation.Target;
 
 import io.javaoperatorsdk.operator.api.reconciler.Constants;
+import io.javaoperatorsdk.operator.api.reconciler.ResourceDiscriminator;
+import io.javaoperatorsdk.operator.processing.event.source.filter.*;
 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;
@@ -68,4 +70,6 @@
    *         itself if no value is set
    */
   Class<? extends GenericFilter> genericFilter() default GenericFilter.class;
+
+  Class<? extends ResourceDiscriminator> resourceDiscriminator() default ResourceDiscriminator.class;
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java
index 930e5fd5b4..328a061e6b 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java
@@ -1,16 +1,13 @@
 package io.javaoperatorsdk.operator.processing.dependent.kubernetes;
 
 import java.util.HashMap;
-import java.util.Optional;
 import java.util.Set;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.fabric8.kubernetes.api.model.KubernetesResourceList;
 import io.fabric8.kubernetes.client.KubernetesClient;
-import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation;
 import io.fabric8.kubernetes.client.dsl.Resource;
 import io.javaoperatorsdk.operator.OperatorException;
 import io.javaoperatorsdk.operator.api.config.informer.InformerConfiguration;
@@ -41,13 +38,12 @@ public abstract class KubernetesDependentResource<R extends HasMetadata, P exten
   protected KubernetesClient client;
   private final Matcher<R, P> matcher;
   private final ResourceUpdatePreProcessor<R> processor;
-  private final Class<R> resourceType;
   private final boolean garbageCollected = this instanceof GarbageCollected;
   private KubernetesDependentResourceConfig kubernetesDependentResourceConfig;
 
   @SuppressWarnings("unchecked")
   public KubernetesDependentResource(Class<R> resourceType) {
-    this.resourceType = resourceType;
+    super(resourceType);
     matcher = this instanceof Matcher ? (Matcher<R, P>) this
         : GenericKubernetesResourceMatcher.matcherFor(resourceType, this);
 
@@ -75,6 +71,7 @@ private void configureWith(String labelSelector, Set<String> namespaces,
         .build();
 
     configureWith(new InformerEventSource<>(ic, context));
+
   }
 
   @SuppressWarnings("unchecked")
@@ -94,7 +91,7 @@ private SecondaryToPrimaryMapper<R> getSecondaryToPrimaryMapper() {
   /**
    * Use to share informers between event more resources.
    *
-   * @param informerEventSource informer to use*
+   * @param informerEventSource informer to use
    */
   public void configureWith(InformerEventSource<R, P> informerEventSource) {
     setEventSource(informerEventSource);
@@ -125,12 +122,12 @@ protected R handleUpdate(R actual, R desired, P primary, Context<P> context) {
 
   @SuppressWarnings("unused")
   public R create(R target, P primary, Context<P> context) {
-    return prepare(target, primary, "Creating").create(target);
+    return prepare(target, primary, "Creating").create();
   }
 
   public R update(R actual, R target, P primary, Context<P> context) {
     var updatedActual = processor.replaceSpecOnActual(actual, target, context);
-    return prepare(target, primary, "Updating").replace(updatedActual);
+    return prepare(updatedActual, primary, "Updating").replace();
   }
 
   public Result<R> match(R actualResource, P primary, Context<P> context) {
@@ -138,13 +135,11 @@ public Result<R> match(R actualResource, P primary, Context<P> context) {
   }
 
   public void delete(P primary, Context<P> context) {
-    var resource = getSecondaryResource(primary);
-    resource.ifPresent(r -> client.resource(r).delete());
+    getSecondaryResource(primary, context).ifPresent(r -> client.resource(r).delete());
   }
 
   @SuppressWarnings("unchecked")
-  protected NonNamespaceOperation<R, KubernetesResourceList<R>, Resource<R>> prepare(R desired,
-      P primary, String actionName) {
+  protected Resource<R> prepare(R desired, P primary, String actionName) {
     log.debug("{} target resource with type: {}, with id: {}",
         actionName,
         desired.getClass(),
@@ -155,7 +150,8 @@ protected NonNamespaceOperation<R, KubernetesResourceList<R>, Resource<R>> prepa
       addDefaultSecondaryToPrimaryMapperAnnotations(desired, primary);
     }
     Class<R> targetClass = (Class<R>) desired.getClass();
-    return client.resources(targetClass).inNamespace(desired.getMetadata().getNamespace());
+    return client.resources(targetClass).inNamespace(desired.getMetadata().getNamespace())
+        .resource(desired);
   }
 
   @Override
@@ -167,6 +163,7 @@ protected InformerEventSource<R, P> createEventSource(EventSourceContext<P> cont
       onUpdateFilter = kubernetesDependentResourceConfig.onUpdateFilter();
       onDeleteFilter = kubernetesDependentResourceConfig.onDeleteFilter();
       genericFilter = kubernetesDependentResourceConfig.genericFilter();
+      setResourceDiscriminator(kubernetesDependentResourceConfig.getResourceDiscriminator());
 
       configureWith(kubernetesDependentResourceConfig.labelSelector(),
           kubernetesDependentResourceConfig.namespaces(),
@@ -203,16 +200,6 @@ protected boolean addOwnerReference() {
     return garbageCollected;
   }
 
-  @Override
-  public Class<R> resourceType() {
-    return resourceType;
-  }
-
-  @Override
-  public Optional<R> getSecondaryResource(P primaryResource) {
-    return eventSource().getSecondaryResource(primaryResource);
-  }
-
   @Override
   public void setKubernetesClient(KubernetesClient kubernetesClient) {
     this.client = kubernetesClient;
@@ -235,5 +222,4 @@ private void prepareEventFiltering(R desired, ResourceID resourceID) {
   private void cleanupAfterEventFiltering(ResourceID resourceID) {
     eventSource().cleanupOnCreateOrUpdateEventFiltering(resourceID);
   }
-
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResourceConfig.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResourceConfig.java
index e4a75b165a..e2a2c0f684 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResourceConfig.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResourceConfig.java
@@ -3,6 +3,7 @@
 import java.util.Set;
 
 import io.javaoperatorsdk.operator.api.reconciler.Constants;
+import io.javaoperatorsdk.operator.api.reconciler.ResourceDiscriminator;
 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;
@@ -15,7 +16,7 @@ public class KubernetesDependentResourceConfig<R> {
   private Set<String> namespaces = Constants.SAME_AS_CONTROLLER_NAMESPACES_SET;
   private String labelSelector = NO_VALUE_SET;
   private boolean namespacesWereConfigured = false;
-
+  private ResourceDiscriminator<R, ?> resourceDiscriminator;
 
   private OnAddFilter<R> onAddFilter;
 
@@ -27,9 +28,9 @@ public class KubernetesDependentResourceConfig<R> {
 
   public KubernetesDependentResourceConfig() {}
 
-  @SuppressWarnings("rawtypes")
   public KubernetesDependentResourceConfig(Set<String> namespaces, String labelSelector,
-      boolean configuredNS, OnAddFilter<R> onAddFilter,
+      boolean configuredNS, ResourceDiscriminator<R, ?> resourceDiscriminator,
+      OnAddFilter<R> onAddFilter,
       OnUpdateFilter<R> onUpdateFilter,
       OnDeleteFilter<R> onDeleteFilter, GenericFilter<R> genericFilter) {
     this.namespaces = namespaces;
@@ -39,10 +40,11 @@ public KubernetesDependentResourceConfig(Set<String> namespaces, String labelSel
     this.onUpdateFilter = onUpdateFilter;
     this.onDeleteFilter = onDeleteFilter;
     this.genericFilter = genericFilter;
+    this.resourceDiscriminator = resourceDiscriminator;
   }
 
   public KubernetesDependentResourceConfig(Set<String> namespaces, String labelSelector) {
-    this(namespaces, labelSelector, true, null, null, null, null);
+    this(namespaces, labelSelector, true, null, null, null, null, null);
   }
 
   public KubernetesDependentResourceConfig<R> setNamespaces(Set<String> namespaces) {
@@ -73,17 +75,21 @@ public OnAddFilter onAddFilter() {
     return onAddFilter;
   }
 
-  @SuppressWarnings("rawtypes")
-  public OnUpdateFilter onUpdateFilter() {
+
+  public OnUpdateFilter<R> onUpdateFilter() {
     return onUpdateFilter;
   }
 
-  @SuppressWarnings("rawtypes")
-  public OnDeleteFilter onDeleteFilter() {
+  public OnDeleteFilter<R> onDeleteFilter() {
     return onDeleteFilter;
   }
 
   public GenericFilter<R> genericFilter() {
     return genericFilter;
   }
+
+  @SuppressWarnings("rawtypes")
+  public ResourceDiscriminator getResourceDiscriminator() {
+    return resourceDiscriminator;
+  }
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/DependentResourceNode.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/DependentResourceNode.java
index e317145916..32cef6c68e 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/DependentResourceNode.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/DependentResourceNode.java
@@ -5,6 +5,7 @@
 import java.util.Optional;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
 
 @SuppressWarnings("rawtypes")
@@ -85,4 +86,8 @@ public DependentResourceNode<R, P> setReadyPostcondition(Condition<R, P> readyPo
   public List<DependentResourceNode> getParents() {
     return parents;
   }
+
+  protected R getSecondaryResource(P primary, Context<P> context) {
+    return getDependentResource().getSecondaryResource(primary, context).orElse(null);
+  }
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowCleanupExecutor.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowCleanupExecutor.java
index 98c6789869..45541b91d6 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowCleanupExecutor.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowCleanupExecutor.java
@@ -104,12 +104,10 @@ public void run() {
           ((Deleter<P>) dependentResourceNode.getDependentResource()).delete(primary, context);
           deleteCalled.add(dependentResourceNode);
         }
-        boolean deletePostConditionMet =
-            deletePostCondition.map(c -> c.isMet(primary,
-                dependentResourceNode.getDependentResource().getSecondaryResource(primary)
-                    .orElse(null),
-                context)).orElse(true);
-
+        boolean deletePostConditionMet = deletePostCondition
+            .map(c -> c.isMet(primary, dependentResourceNode.getSecondaryResource(primary, context),
+                context))
+            .orElse(true);
         if (deletePostConditionMet) {
           alreadyVisited.add(dependentResourceNode);
           handleDependentCleaned(dependentResourceNode);
@@ -127,6 +125,7 @@ public void run() {
     }
   }
 
+
   private synchronized void handleDependentCleaned(
       DependentResourceNode<?, P> dependentResourceNode) {
     var dependOns = dependentResourceNode.getDependsOn();
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutor.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutor.java
index 14028cb980..33ecd12f77 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutor.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutor.java
@@ -85,8 +85,7 @@ private synchronized <R> void handleReconcile(DependentResourceNode<R, P> depend
     }
 
     boolean reconcileConditionMet = dependentResourceNode.getReconcilePrecondition()
-        .map(rc -> rc.isMet(primary,
-            dependentResourceNode.getDependentResource().getSecondaryResource(primary).orElse(null),
+        .map(rc -> rc.isMet(primary, dependentResourceNode.getSecondaryResource(primary, context),
             context))
         .orElse(true);
 
@@ -167,9 +166,12 @@ public void run() {
         ReconcileResult reconcileResult = dependentResource.reconcile(primary, context);
         reconcileResults.put(dependentResource, reconcileResult);
         reconciled.add(dependentResourceNode);
+
         boolean ready = dependentResourceNode.getReadyPostcondition()
             .map(rc -> rc.isMet(primary,
-                dependentResourceNode.getDependentResource().getSecondaryResource(primary)
+                context
+                    .getSecondaryResource(
+                        dependentResourceNode.getDependentResource().resourceType())
                     .orElse(null),
                 context))
             .orElse(true);
@@ -210,8 +212,8 @@ public void run() {
         }
         boolean deletePostConditionMet =
             deletePostCondition.map(c -> c.isMet(primary,
-                dependentResourceNode.getDependentResource().getSecondaryResource(primary)
-                    .orElse(null),
+                context.getSecondaryResource(
+                    dependentResourceNode.getDependentResource().resourceType()).orElse(null),
                 context)).orElse(true);
         if (deletePostConditionMet) {
           alreadyVisited.add(dependentResourceNode);
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java
index 58bf5ee00a..e0280796b2 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java
@@ -135,7 +135,8 @@ private void handleMarkedEventForResource(ResourceState state) {
   private void submitReconciliationExecution(ResourceState state) {
     try {
       boolean controllerUnderExecution = isControllerUnderExecution(state);
-      Optional<R> maybeLatest = cache.get(state.getId());
+      final var resourceID = state.getId();
+      Optional<R> maybeLatest = cache.get(resourceID);
       maybeLatest.ifPresent(MDCUtils::addResourceInfo);
       if (!controllerUnderExecution && maybeLatest.isPresent()) {
         var rateLimit = state.getRateLimit();
@@ -145,24 +146,24 @@ private void submitReconciliationExecution(ResourceState state) {
         }
         var rateLimiterPermission = rateLimiter.isLimited(rateLimit);
         if (rateLimiterPermission.isPresent()) {
-          handleRateLimitedSubmission(state.getId(), rateLimiterPermission.get());
+          handleRateLimitedSubmission(resourceID, rateLimiterPermission.get());
           return;
         }
         state.setUnderProcessing(true);
         final var latest = maybeLatest.get();
         ExecutionScope<R> executionScope = new ExecutionScope<>(latest, state.getRetry());
         state.unMarkEventReceived();
-        metrics.reconcileCustomResource(state.getId(), state.getRetry(), metricsMetadata);
+        metrics.reconcileCustomResource(resourceID, state.getRetry(), metricsMetadata);
         log.debug("Executing events for custom resource. Scope: {}", executionScope);
         executor.execute(new ReconcilerExecutor(executionScope));
       } else {
         log.debug(
             "Skipping executing controller for resource id: {}. Controller in execution: {}. Latest Resource present: {}",
-            state,
+            resourceID,
             controllerUnderExecution,
             maybeLatest.isPresent());
         if (maybeLatest.isEmpty()) {
-          log.debug("no custom resource found in cache for ResourceID: {}", state);
+          log.debug("no custom resource found in cache for resource id: {}", resourceID);
         }
       }
     } finally {
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSourceManager.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSourceManager.java
index 3f1e5e848d..fc68e6e413 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSourceManager.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSourceManager.java
@@ -23,7 +23,8 @@
 import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceAction;
 import io.javaoperatorsdk.operator.processing.event.source.timer.TimerEventSource;
 
-public class EventSourceManager<P extends HasMetadata> implements LifecycleAware {
+public class EventSourceManager<P extends HasMetadata>
+    implements LifecycleAware, EventSourceRetriever<P> {
 
   private static final Logger log = LoggerFactory.getLogger(EventSourceManager.class);
 
@@ -171,17 +172,23 @@ public ControllerResourceEventSource<P> getControllerResourceEventSource() {
     return eventSources.controllerResourceEventSource();
   }
 
-  <S> ResourceEventSource<S, P> getResourceEventSourceFor(
-      Class<S> dependentType) {
+  @Override
+  public <R> ResourceEventSource<R, P> getResourceEventSourceFor(Class<R> dependentType) {
     return getResourceEventSourceFor(dependentType, null);
   }
 
-  public <S> List<ResourceEventSource<S, P>> getEventSourcesFor(Class<S> dependentType) {
+  public <R> List<ResourceEventSource<R, P>> getResourceEventSourcesFor(Class<R> dependentType) {
     return eventSources.getEventSources(dependentType);
   }
 
-  public <S> ResourceEventSource<S, P> getResourceEventSourceFor(
-      Class<S> dependentType, String qualifier) {
+  @Deprecated
+  public <R> List<ResourceEventSource<R, P>> getEventSourcesFor(Class<R> dependentType) {
+    return eventSources.getEventSources(dependentType);
+  }
+
+  @Override
+  public <R> ResourceEventSource<R, P> getResourceEventSourceFor(
+      Class<R> dependentType, String qualifier) {
     Objects.requireNonNull(dependentType, "dependentType is Mandatory");
     return eventSources.get(dependentType, qualifier);
   }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSourceRetriever.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSourceRetriever.java
new file mode 100644
index 0000000000..a31d8902b0
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSourceRetriever.java
@@ -0,0 +1,18 @@
+package io.javaoperatorsdk.operator.processing.event;
+
+import java.util.List;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.processing.event.source.ResourceEventSource;
+
+public interface EventSourceRetriever<P extends HasMetadata> {
+
+  <R> ResourceEventSource<R, P> getResourceEventSourceFor(
+      Class<R> dependentType);
+
+  <R> ResourceEventSource<R, P> getResourceEventSourceFor(
+      Class<R> dependentType, String qualifier);
+
+  <R> List<ResourceEventSource<R, P>> getResourceEventSourcesFor(Class<R> dependentType);
+
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSources.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSources.java
index 17de7ff947..e4fabe7ff8 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSources.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSources.java
@@ -8,7 +8,6 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.processing.Controller;
-import io.javaoperatorsdk.operator.processing.ResourceOwner;
 import io.javaoperatorsdk.operator.processing.event.source.EventSource;
 import io.javaoperatorsdk.operator.processing.event.source.ResourceEventSource;
 import io.javaoperatorsdk.operator.processing.event.source.controller.ControllerResourceEventSource;
@@ -86,8 +85,8 @@ public void add(String name, EventSource eventSource) {
 
   @SuppressWarnings("rawtypes")
   private Class<?> getResourceType(EventSource source) {
-    return source instanceof ResourceOwner
-        ? ((ResourceOwner) source).resourceType()
+    return source instanceof ResourceEventSource
+        ? ((ResourceEventSource) source).resourceType()
         : source.getClass();
   }
 
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ResourceID.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ResourceID.java
index 7baeab6a4b..b6c2d976e3 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ResourceID.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ResourceID.java
@@ -54,6 +54,12 @@ public boolean equals(Object o) {
         that.namespace);
   }
 
+  public boolean isSameResource(HasMetadata hasMetadata) {
+    final var metadata = hasMetadata.getMetadata();
+    return getName().equals(metadata.getName()) &&
+        getNamespace().map(ns -> ns.equals(metadata.getNamespace())).orElse(true);
+  }
+
   @Override
   public int hashCode() {
     return Objects.hash(name, namespace);
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ResourceEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ResourceEventSource.java
index bf6b9c0fd3..38b5d7007f 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ResourceEventSource.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ResourceEventSource.java
@@ -4,14 +4,19 @@
 import java.util.Set;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.javaoperatorsdk.operator.processing.ResourceOwner;
 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;
 
-public interface ResourceEventSource<R, P extends HasMetadata> extends EventSource,
-    ResourceOwner<R, P> {
+public interface ResourceEventSource<R, P extends HasMetadata> extends EventSource {
+
+  /**
+   * Retrieves the resource type associated with this ResourceEventSource
+   *
+   * @return the resource type associated with this ResourceEventSource
+   */
+  Class<R> resourceType();
 
   default Optional<R> getSecondaryResource(P primary) {
     var resources = getSecondaryResources(primary);
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/ManagedInformerEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/ManagedInformerEventSource.java
index 6ebd63a7eb..cc9af59094 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/ManagedInformerEventSource.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/ManagedInformerEventSource.java
@@ -93,10 +93,10 @@ public void handleRecentResourceCreate(ResourceID resourceID, R resource) {
   public Optional<R> get(ResourceID resourceID) {
     Optional<R> resource = temporaryResourceCache.getResourceFromCache(resourceID);
     if (resource.isPresent()) {
-      log.debug("Resource found in temporal cache for Resource ID: {}", resourceID);
+      log.debug("Resource found in temporary cache for Resource ID: {}", resourceID);
       return resource;
     } else {
-      log.debug("Resource not found in temporal cache reading it from informer cache," +
+      log.debug("Resource not found in temporary cache reading it from informer cache," +
           " for Resource ID: {}", resourceID);
       return cache.get(resourceID);
     }
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverriderTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverriderTest.java
index 127cd535b1..6e486b5347 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverriderTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverriderTest.java
@@ -1,6 +1,5 @@
 package io.javaoperatorsdk.operator.api.config;
 
-import java.util.Optional;
 import java.util.Set;
 
 import org.junit.jupiter.api.Test;
@@ -85,10 +84,6 @@ public Class<Object> resourceType() {
         return Object.class;
       }
 
-      @Override
-      public Optional<Object> getSecondaryResource(ConfigMap primary) {
-        return Optional.empty();
-      }
     }
   }
 
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/AbstractDependentResourceTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/AbstractDependentResourceTest.java
index 4edd3dfb4c..b93abd45c0 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/AbstractDependentResourceTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/AbstractDependentResourceTest.java
@@ -16,7 +16,6 @@
 
 class AbstractDependentResourceTest {
 
-
   @Test
   void throwsExceptionIfDesiredIsNullOnCreate() {
     TestDependentResource testDependentResource = new TestDependentResource();
@@ -80,7 +79,8 @@ public Class<ConfigMap> resourceType() {
     }
 
     @Override
-    public Optional<ConfigMap> getSecondaryResource(TestCustomResource primary) {
+    public Optional<ConfigMap> getSecondaryResource(TestCustomResource primary,
+        Context<TestCustomResource> context) {
       return Optional.ofNullable(secondary);
     }
 
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/EmptyTestDependentResource.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/EmptyTestDependentResource.java
index aa75849051..dab1bc0132 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/EmptyTestDependentResource.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/EmptyTestDependentResource.java
@@ -1,7 +1,5 @@
 package io.javaoperatorsdk.operator.processing.dependent;
 
-import java.util.Optional;
-
 import io.fabric8.kubernetes.api.model.apps.Deployment;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
@@ -17,11 +15,6 @@ public ReconcileResult<Deployment> reconcile(TestCustomResource primary,
     return null;
   }
 
-  @Override
-  public Optional<Deployment> getSecondaryResource(TestCustomResource primaryResource) {
-    return Optional.empty();
-  }
-
   @Override
   public Class<Deployment> resourceType() {
     return Deployment.class;
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractSimpleDependentResourceTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractSimpleDependentResourceTest.java
index e020fd27a3..520f44365f 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractSimpleDependentResourceTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractSimpleDependentResourceTest.java
@@ -45,14 +45,14 @@ void getsTheResourceFromSupplyIfReconciling() {
     simpleDependentResource.reconcile(TestUtils.testCustomResource1(), null);
 
     verify(supplierMock, times(1)).get();
-    assertThat(simpleDependentResource.getSecondaryResource(TestUtils.testCustomResource1()))
+    assertThat(simpleDependentResource.getSecondaryResource(TestUtils.testCustomResource1(), null))
         .isPresent()
         .isEqualTo(Optional.of(SampleExternalResource.testResource1()));
   }
 
   @Test
   void getResourceReadsTheResourceFromCache() {
-    simpleDependentResource.getSecondaryResource(TestUtils.testCustomResource1());
+    simpleDependentResource.getSecondaryResource(TestUtils.testCustomResource1(), null);
 
     verify(supplierMock, times(0)).get();
     verify(updatableCacheMock, times(1)).get(any());
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutorTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutorTest.java
index 8dfab3fa20..9c10c06cc0 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutorTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutorTest.java
@@ -3,7 +3,6 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
-import java.util.Optional;
 
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.Deleter;
@@ -49,11 +48,6 @@ public Class<String> resourceType() {
       return String.class;
     }
 
-    @Override
-    public Optional<String> getSecondaryResource(TestCustomResource primary) {
-      return Optional.of(VALUE);
-    }
-
     @Override
     public String toString() {
       return name;
@@ -113,11 +107,6 @@ public Class<String> resourceType() {
       return String.class;
     }
 
-    @Override
-    public Optional<String> getSecondaryResource(TestCustomResource primary) {
-      return Optional.of(VALUE);
-    }
-
     @Override
     public String toString() {
       return name;
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowCleanupExecutorTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowCleanupExecutorTest.java
index 7c1c5d6ff6..56bd876687 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowCleanupExecutorTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowCleanupExecutorTest.java
@@ -4,17 +4,20 @@
 import org.junit.jupiter.api.Test;
 
 import io.javaoperatorsdk.operator.AggregatedOperatorException;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.processing.dependent.workflow.builder.WorkflowBuilder;
 import io.javaoperatorsdk.operator.sample.simple.TestCustomResource;
 
 import static io.javaoperatorsdk.operator.processing.dependent.workflow.ExecutionAssert.assertThat;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
 
 class WorkflowCleanupExecutorTest extends AbstractWorkflowExecutorTest {
 
   protected TestDeleterDependent dd1 = new TestDeleterDependent("DR_DELETER_1");
   protected TestDeleterDependent dd2 = new TestDeleterDependent("DR_DELETER_2");
   protected TestDeleterDependent dd3 = new TestDeleterDependent("DR_DELETER_3");
+  Context<TestCustomResource> mockContext = mock(Context.class);
 
   @Test
   void cleanUpDiamondWorkflow() {
@@ -45,7 +48,7 @@ void dontDeleteIfDependentErrored() {
         .withThrowExceptionFurther(false)
         .build();
 
-    var res = workflow.cleanup(new TestCustomResource(), null);
+    var res = workflow.cleanup(new TestCustomResource(), mockContext);
     assertThrows(AggregatedOperatorException.class,
         res::throwAggregateExceptionIfErrorsPresent);
 
@@ -64,7 +67,7 @@ void cleanupConditionTrivialCase() {
         .addDependentResource(dd2).dependsOn(dd1).withDeletePostcondition(noMetDeletePostCondition)
         .build();
 
-    var res = workflow.cleanup(new TestCustomResource(), null);
+    var res = workflow.cleanup(new TestCustomResource(), mockContext);
 
     assertThat(executionHistory).deleted(dd2).notReconciled(dd1);
     Assertions.assertThat(res.getDeleteCalledOnDependents()).containsExactlyInAnyOrder(dd2);
@@ -79,7 +82,7 @@ void cleanupConditionMet() {
         .addDependentResource(dd2).dependsOn(dd1).withDeletePostcondition(metDeletePostCondition)
         .build();
 
-    var res = workflow.cleanup(new TestCustomResource(), null);
+    var res = workflow.cleanup(new TestCustomResource(), mockContext);
 
     assertThat(executionHistory).deleted(dd2, dd1);
 
@@ -99,7 +102,7 @@ void cleanupConditionDiamondWorkflow() {
         .addDependentResource(dd4).dependsOn(dd2, dd3)
         .build();
 
-    var res = workflow.cleanup(new TestCustomResource(), null);
+    var res = workflow.cleanup(new TestCustomResource(), mockContext);
 
     assertThat(executionHistory)
         .reconciledInOrder(dd4, dd2)
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutorTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutorTest.java
index fa9d757b67..873cf66cb6 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutorTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutorTest.java
@@ -4,11 +4,13 @@
 import org.junit.jupiter.api.Test;
 
 import io.javaoperatorsdk.operator.AggregatedOperatorException;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.processing.dependent.workflow.builder.WorkflowBuilder;
 import io.javaoperatorsdk.operator.sample.simple.TestCustomResource;
 
 import static io.javaoperatorsdk.operator.processing.dependent.workflow.ExecutionAssert.assertThat;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
 
 @SuppressWarnings("rawtypes")
 class WorkflowReconcileExecutorTest extends AbstractWorkflowExecutorTest {
@@ -21,6 +23,8 @@ class WorkflowReconcileExecutorTest extends AbstractWorkflowExecutorTest {
   private final Condition<String, TestCustomResource> notMetReadyCondition =
       (primary, secondary, context) -> false;
 
+  Context<TestCustomResource> mockContext = mock(Context.class);
+
   @Test
   void reconcileTopLevelResources() {
     var workflow = new WorkflowBuilder<TestCustomResource>()
@@ -28,7 +32,7 @@ void reconcileTopLevelResources() {
         .addDependentResource(dr2)
         .build();
 
-    var res = workflow.reconcile(new TestCustomResource(), null);
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
 
     assertThat(executionHistory).reconciled(dr1, dr2);
     Assertions.assertThat(res.getErroredDependents()).isEmpty();
@@ -42,7 +46,7 @@ void reconciliationWithSimpleDependsOn() {
         .addDependentResource(dr2).dependsOn(dr1)
         .build();
 
-    var res = workflow.reconcile(new TestCustomResource(), null);
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
 
     Assertions.assertThat(res.getErroredDependents()).isEmpty();
     assertThat(executionHistory).reconciledInOrder(dr1, dr2);
@@ -61,7 +65,7 @@ void reconciliationWithTwoTheDependsOns() {
         .addDependentResource(dr3).dependsOn(dr1)
         .build();
 
-    var res = workflow.reconcile(new TestCustomResource(), null);
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
 
     Assertions.assertThat(res.getErroredDependents()).isEmpty();
     assertThat(executionHistory)
@@ -83,7 +87,7 @@ void diamondShareWorkflowReconcile() {
         .addDependentResource(dr4).dependsOn(dr3).dependsOn(dr2)
         .build();
 
-    var res = workflow.reconcile(new TestCustomResource(), null);
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
 
     Assertions.assertThat(res.getErroredDependents()).isEmpty();
     assertThat(executionHistory)
@@ -103,7 +107,7 @@ void exceptionHandlingSimpleCases() {
         .withThrowExceptionFurther(false)
         .build();
 
-    var res = workflow.reconcile(new TestCustomResource(), null);
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
 
     assertThrows(AggregatedOperatorException.class,
         res::throwAggregateExceptionIfErrorsPresent);
@@ -123,7 +127,7 @@ void dependentsOnErroredResourceNotReconciled() {
         .withThrowExceptionFurther(false)
         .build();
 
-    var res = workflow.reconcile(new TestCustomResource(), null);
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
     assertThrows(AggregatedOperatorException.class,
         res::throwAggregateExceptionIfErrorsPresent);
 
@@ -145,7 +149,7 @@ void oneBranchErrorsOtherCompletes() {
         .withThrowExceptionFurther(false)
         .build();
 
-    var res = workflow.reconcile(new TestCustomResource(), null);
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
     assertThrows(AggregatedOperatorException.class,
         res::throwAggregateExceptionIfErrorsPresent);
 
@@ -164,7 +168,7 @@ void onlyOneDependsOnErroredResourceNotReconciled() {
         .withThrowExceptionFurther(false)
         .build();
 
-    var res = workflow.reconcile(new TestCustomResource(), null);
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
     assertThrows(AggregatedOperatorException.class,
         res::throwAggregateExceptionIfErrorsPresent);
 
@@ -182,7 +186,7 @@ void simpleReconcileCondition() {
         .addDependentResource(drDeleter).withReconcilePrecondition(not_met_reconcile_condition)
         .build();
 
-    var res = workflow.reconcile(new TestCustomResource(), null);
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
 
     assertThat(executionHistory).notReconciled(dr1).reconciled(dr2).deleted(drDeleter);
     Assertions.assertThat(res.getErroredDependents()).isEmpty();
@@ -200,7 +204,7 @@ void triangleOnceConditionNotMet() {
         .dependsOn(dr1)
         .build();
 
-    var res = workflow.reconcile(new TestCustomResource(), null);
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
 
     assertThat(executionHistory).reconciledInOrder(dr1, dr2).deleted(drDeleter);
     Assertions.assertThat(res.getErroredDependents()).isEmpty();
@@ -222,7 +226,7 @@ void reconcileConditionTransitiveDelete() {
         .withReconcilePrecondition(met_reconcile_condition)
         .build();
 
-    var res = workflow.reconcile(new TestCustomResource(), null);
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
 
     Assertions.assertThat(res.getErroredDependents()).isEmpty();
     assertThat(executionHistory).notReconciled(dr2);
@@ -246,7 +250,7 @@ void reconcileConditionAlsoErrorDependsOn() {
         .withThrowExceptionFurther(false)
         .build();
 
-    var res = workflow.reconcile(new TestCustomResource(), null);
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
     assertThrows(AggregatedOperatorException.class,
         res::throwAggregateExceptionIfErrorsPresent);
 
@@ -267,7 +271,7 @@ void oneDependsOnConditionNotMet() {
         .addDependentResource(drDeleter).dependsOn(dr1, dr2)
         .build();
 
-    var res = workflow.reconcile(new TestCustomResource(), null);
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
 
     Assertions.assertThat(res.getErroredDependents()).isEmpty();
 
@@ -287,7 +291,7 @@ void deletedIfReconcileConditionNotMet() {
         .addDependentResource(drDeleter2).dependsOn(dr1, drDeleter)
         .build();
 
-    var res = workflow.reconcile(new TestCustomResource(), null);
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
 
     assertThat(executionHistory)
         .reconciledInOrder(dr1, drDeleter2, drDeleter)
@@ -313,7 +317,7 @@ void deleteDoneInReverseOrder() {
         .addDependentResource(drDeleter4).dependsOn(drDeleter3)
         .build();
 
-    var res = workflow.reconcile(new TestCustomResource(), null);
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
 
     assertThat(executionHistory)
         .reconciledInOrder(dr1, drDeleter4, drDeleter3, drDeleter)
@@ -339,7 +343,7 @@ void diamondDeleteWithPostConditionInMiddle() {
         .addDependentResource(drDeleter4).dependsOn(drDeleter3, drDeleter2)
         .build();
 
-    var res = workflow.reconcile(new TestCustomResource(), null);
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
 
     assertThat(executionHistory).notReconciled(drDeleter)
         .reconciledInOrder(drDeleter4, drDeleter2)
@@ -363,7 +367,7 @@ void diamondDeleteErrorInMiddle() {
         .withThrowExceptionFurther(false)
         .build();
 
-    var res = workflow.reconcile(new TestCustomResource(), null);
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
 
     assertThat(executionHistory)
         .notReconciled(drDeleter, drError)
@@ -381,7 +385,7 @@ void readyConditionTrivialCase() {
         .addDependentResource(dr2).dependsOn(dr1)
         .build();
 
-    var res = workflow.reconcile(new TestCustomResource(), null);
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
 
     assertThat(executionHistory).reconciledInOrder(dr1, dr2);
 
@@ -397,7 +401,7 @@ void readyConditionNotMetTrivialCase() {
         .addDependentResource(dr2).dependsOn(dr1)
         .build();
 
-    var res = workflow.reconcile(new TestCustomResource(), null);
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
 
 
     assertThat(executionHistory).reconciled(dr1).notReconciled(dr2);
@@ -417,7 +421,7 @@ void readyConditionNotMetInOneParent() {
         .addDependentResource(dr3).dependsOn(dr1, dr2)
         .build();
 
-    var res = workflow.reconcile(new TestCustomResource(), null);
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
 
     assertThat(executionHistory).reconciled(dr1, dr2).notReconciled(dr3);
     Assertions.assertThat(res.getErroredDependents()).isEmpty();
@@ -437,7 +441,7 @@ void diamondShareWithReadyCondition() {
         .addDependentResource(dr4).dependsOn(dr2, dr3)
         .build();
 
-    var res = workflow.reconcile(new TestCustomResource(), null);
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
 
     Assertions.assertThat(res.getErroredDependents()).isEmpty();
     assertThat(executionHistory).reconciledInOrder(dr1, dr2)
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/IndexDiscriminatorIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/IndexDiscriminatorIT.java
new file mode 100644
index 0000000000..fe5b63de8a
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/IndexDiscriminatorIT.java
@@ -0,0 +1,77 @@
+package io.javaoperatorsdk.operator;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension;
+import io.javaoperatorsdk.operator.sample.indexdiscriminator.IndexDiscriminatorTestCustomResource;
+import io.javaoperatorsdk.operator.sample.indexdiscriminator.IndexDiscriminatorTestReconciler;
+import io.javaoperatorsdk.operator.sample.indexdiscriminator.IndexDiscriminatorTestSpec;
+
+import static io.javaoperatorsdk.operator.sample.indexdiscriminator.IndexDiscriminatorTestDRConfigMap.DATA_KEY;
+import static io.javaoperatorsdk.operator.sample.indexdiscriminator.IndexDiscriminatorTestReconciler.FIRST_CONFIG_MAP_SUFFIX_1;
+import static io.javaoperatorsdk.operator.sample.indexdiscriminator.IndexDiscriminatorTestReconciler.FIRST_CONFIG_MAP_SUFFIX_2;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+class IndexDiscriminatorIT {
+
+  public static final String TEST_RESOURCE_1 = "test1";
+  public static final String CHANGED_SPEC_VALUE = "otherValue";
+  @RegisterExtension
+  LocallyRunOperatorExtension operator =
+      LocallyRunOperatorExtension.builder().withReconciler(IndexDiscriminatorTestReconciler.class)
+          .build();
+
+  @Test
+  void resourcesFoundAndReconciled() {
+    var res = operator.create(createTestCustomResource());
+    var reconciler = operator.getReconcilerOfType(IndexDiscriminatorTestReconciler.class);
+
+    await().untilAsserted(() -> {
+      assertThat(reconciler.getNumberOfExecutions()).isEqualTo(1);
+      assertThat(operator.get(ConfigMap.class, TEST_RESOURCE_1 + FIRST_CONFIG_MAP_SUFFIX_1))
+          .isNotNull();
+      assertThat(operator.get(ConfigMap.class, TEST_RESOURCE_1 + FIRST_CONFIG_MAP_SUFFIX_2))
+          .isNotNull();
+    });
+
+    res.getSpec().setValue(CHANGED_SPEC_VALUE);
+    res = operator.replace(res);
+
+    await().untilAsserted(() -> {
+      assertThat(reconciler.getNumberOfExecutions()).isEqualTo(2);
+      var cm1 = operator.get(ConfigMap.class, TEST_RESOURCE_1 + FIRST_CONFIG_MAP_SUFFIX_1);
+      var cm2 = operator.get(ConfigMap.class, TEST_RESOURCE_1 + FIRST_CONFIG_MAP_SUFFIX_2);
+      assertThat(cm1).isNotNull();
+      assertThat(cm2).isNotNull();
+      assertThat(cm1.getData().get(DATA_KEY)).isEqualTo(CHANGED_SPEC_VALUE);
+      assertThat(cm2.getData().get(DATA_KEY)).isEqualTo(CHANGED_SPEC_VALUE);
+    });
+
+    operator.delete(res);
+
+    await().untilAsserted(() -> {
+      var cm1 = operator.get(ConfigMap.class, TEST_RESOURCE_1 + FIRST_CONFIG_MAP_SUFFIX_1);
+      var cm2 = operator.get(ConfigMap.class, TEST_RESOURCE_1 + FIRST_CONFIG_MAP_SUFFIX_2);
+      assertThat(cm1).isNull();
+      assertThat(cm2).isNull();
+    });
+  }
+
+  public IndexDiscriminatorTestCustomResource createTestCustomResource() {
+    IndexDiscriminatorTestCustomResource resource =
+        new IndexDiscriminatorTestCustomResource();
+    resource.setMetadata(
+        new ObjectMetaBuilder()
+            .withName(TEST_RESOURCE_1)
+            .withNamespace(operator.getNamespace())
+            .build());
+    resource.setSpec(new IndexDiscriminatorTestSpec());
+    resource.getSpec().setValue("default");
+    return resource;
+  }
+
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/indexdiscriminator/IndexDiscriminator.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/indexdiscriminator/IndexDiscriminator.java
new file mode 100644
index 0000000000..eb6e193479
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/indexdiscriminator/IndexDiscriminator.java
@@ -0,0 +1,41 @@
+package io.javaoperatorsdk.operator.sample.indexdiscriminator;
+
+import java.util.Optional;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ResourceDiscriminator;
+import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
+
+import static io.javaoperatorsdk.operator.sample.indexdiscriminator.IndexDiscriminatorTestReconciler.configMapKeyFromPrimary;
+
+public class IndexDiscriminator
+    implements ResourceDiscriminator<ConfigMap, IndexDiscriminatorTestCustomResource> {
+
+  private final String indexName;
+  private final String nameSuffix;
+
+  public IndexDiscriminator(String indexName, String nameSuffix) {
+    this.indexName = indexName;
+    this.nameSuffix = nameSuffix;
+  }
+
+  @Override
+  public Optional<ConfigMap> distinguish(Class<ConfigMap> resource,
+      IndexDiscriminatorTestCustomResource primary,
+      Context<IndexDiscriminatorTestCustomResource> context) {
+
+    InformerEventSource<ConfigMap, IndexDiscriminatorTestCustomResource> eventSource =
+        (InformerEventSource<ConfigMap, IndexDiscriminatorTestCustomResource>) context
+            .eventSourceRetriever()
+            .getResourceEventSourceFor(ConfigMap.class);
+    var resources = eventSource.byIndex(indexName, configMapKeyFromPrimary(primary, nameSuffix));
+    if (resources.isEmpty()) {
+      return Optional.empty();
+    } else if (resources.size() > 1) {
+      throw new IllegalStateException("more than one resource");
+    } else {
+      return Optional.of(resources.get(0));
+    }
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/indexdiscriminator/IndexDiscriminatorTestCustomResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/indexdiscriminator/IndexDiscriminatorTestCustomResource.java
new file mode 100644
index 0000000000..729b1d80eb
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/indexdiscriminator/IndexDiscriminatorTestCustomResource.java
@@ -0,0 +1,16 @@
+package io.javaoperatorsdk.operator.sample.indexdiscriminator;
+
+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("idt")
+public class IndexDiscriminatorTestCustomResource
+    extends CustomResource<IndexDiscriminatorTestSpec, IndexDiscriminatorTestStatus>
+    implements Namespaced {
+
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/indexdiscriminator/IndexDiscriminatorTestDRConfigMap.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/indexdiscriminator/IndexDiscriminatorTestDRConfigMap.java
new file mode 100644
index 0000000000..88dc40f55c
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/indexdiscriminator/IndexDiscriminatorTestDRConfigMap.java
@@ -0,0 +1,38 @@
+package io.javaoperatorsdk.operator.sample.indexdiscriminator;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.processing.dependent.kubernetes.CRUDNoGCKubernetesDependentResource;
+import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependent;
+
+@KubernetesDependent
+public class IndexDiscriminatorTestDRConfigMap
+    extends CRUDNoGCKubernetesDependentResource<ConfigMap, IndexDiscriminatorTestCustomResource> {
+
+  public static final String DATA_KEY = "key";
+  private final String suffix;
+
+  public IndexDiscriminatorTestDRConfigMap(String value) {
+    super(ConfigMap.class);
+    this.suffix = value;
+  }
+
+  @Override
+  protected ConfigMap desired(IndexDiscriminatorTestCustomResource primary,
+      Context<IndexDiscriminatorTestCustomResource> context) {
+    Map<String, String> data = new HashMap<>();
+    data.put(DATA_KEY, primary.getSpec().getValue());
+
+    return new ConfigMapBuilder()
+        .withNewMetadata()
+        .withName(primary.getMetadata().getName() + suffix)
+        .withNamespace(primary.getMetadata().getNamespace())
+        .endMetadata()
+        .withData(data)
+        .build();
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/indexdiscriminator/IndexDiscriminatorTestReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/indexdiscriminator/IndexDiscriminatorTestReconciler.java
new file mode 100644
index 0000000000..0b0af2a1cc
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/indexdiscriminator/IndexDiscriminatorTestReconciler.java
@@ -0,0 +1,120 @@
+package io.javaoperatorsdk.operator.sample.indexdiscriminator;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.javaoperatorsdk.operator.api.config.informer.InformerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.*;
+import io.javaoperatorsdk.operator.junit.KubernetesClientAware;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
+import io.javaoperatorsdk.operator.support.TestExecutionInfoProvider;
+
+@ControllerConfiguration
+public class IndexDiscriminatorTestReconciler
+    implements Reconciler<IndexDiscriminatorTestCustomResource>,
+    Cleaner<IndexDiscriminatorTestCustomResource>,
+    TestExecutionInfoProvider, EventSourceInitializer<IndexDiscriminatorTestCustomResource>,
+    KubernetesClientAware {
+
+  public static final String FIRST_CONFIG_MAP_SUFFIX_1 = "-1";
+  public static final String FIRST_CONFIG_MAP_SUFFIX_2 = "-2";
+  public static final String CONFIG_MAP_INDEX_1 = "CONFIG_MAP_INDEX1";
+  public static final String CONFIG_MAP_INDEX_2 = "CONFIG_MAP_INDEX2";
+
+  private final AtomicInteger numberOfExecutions = new AtomicInteger(0);
+
+  private final IndexDiscriminatorTestDRConfigMap firstDependentResourceConfigMap;
+  private final IndexDiscriminatorTestDRConfigMap secondDependentResourceConfigMap;
+  private KubernetesClient client;
+
+  public IndexDiscriminatorTestReconciler() {
+    firstDependentResourceConfigMap =
+        new IndexDiscriminatorTestDRConfigMap(FIRST_CONFIG_MAP_SUFFIX_1);
+    secondDependentResourceConfigMap =
+        new IndexDiscriminatorTestDRConfigMap(FIRST_CONFIG_MAP_SUFFIX_2);
+  }
+
+  @Override
+  public UpdateControl<IndexDiscriminatorTestCustomResource> reconcile(
+      IndexDiscriminatorTestCustomResource resource,
+      Context<IndexDiscriminatorTestCustomResource> context) {
+    numberOfExecutions.getAndIncrement();
+    firstDependentResourceConfigMap.reconcile(resource, context);
+    secondDependentResourceConfigMap.reconcile(resource, context);
+    return UpdateControl.noUpdate();
+  }
+
+  public int getNumberOfExecutions() {
+    return numberOfExecutions.get();
+  }
+
+  @Override
+  public Map<String, EventSource> prepareEventSources(
+      EventSourceContext<IndexDiscriminatorTestCustomResource> context) {
+
+    InformerEventSource<ConfigMap, IndexDiscriminatorTestCustomResource> eventSource =
+        new InformerEventSource<>(InformerConfiguration.from(ConfigMap.class, context)
+            .build(), context);
+
+    eventSource.addIndexer(CONFIG_MAP_INDEX_1, cm -> {
+      if (cm.getMetadata().getName().endsWith(FIRST_CONFIG_MAP_SUFFIX_1)) {
+        return List.of(configMapKey(cm));
+      } else {
+        return Collections.emptyList();
+      }
+    });
+    eventSource.addIndexer(CONFIG_MAP_INDEX_2, cm -> {
+      if (cm.getMetadata().getName().endsWith(FIRST_CONFIG_MAP_SUFFIX_2)) {
+        return List.of(configMapKey(cm));
+      } else {
+        return Collections.emptyList();
+      }
+    });
+
+    firstDependentResourceConfigMap.configureWith(eventSource);
+    secondDependentResourceConfigMap.configureWith(eventSource);
+
+    firstDependentResourceConfigMap
+        .setResourceDiscriminator(
+            new IndexDiscriminator(CONFIG_MAP_INDEX_1, FIRST_CONFIG_MAP_SUFFIX_1));
+    secondDependentResourceConfigMap
+        .setResourceDiscriminator(
+            new IndexDiscriminator(CONFIG_MAP_INDEX_2, FIRST_CONFIG_MAP_SUFFIX_2));
+    return EventSourceInitializer.nameEventSources(eventSource);
+  }
+
+  @Override
+  public KubernetesClient getKubernetesClient() {
+    return client;
+  }
+
+  @Override
+  public void setKubernetesClient(KubernetesClient kubernetesClient) {
+    this.client = kubernetesClient;
+    firstDependentResourceConfigMap.setKubernetesClient(kubernetesClient);
+    secondDependentResourceConfigMap.setKubernetesClient(kubernetesClient);
+  }
+
+  public static String configMapKey(ConfigMap configMap) {
+    return configMap.getMetadata().getName() + "#" + configMap.getMetadata().getNamespace();
+  }
+
+  public static String configMapKeyFromPrimary(IndexDiscriminatorTestCustomResource primary,
+      String nameSuffix) {
+    return primary.getMetadata().getName() + nameSuffix + "#"
+        + primary.getMetadata().getNamespace();
+  }
+
+  @Override
+  public DeleteControl cleanup(IndexDiscriminatorTestCustomResource resource,
+      Context<IndexDiscriminatorTestCustomResource> context) {
+    firstDependentResourceConfigMap.delete(resource, context);
+    secondDependentResourceConfigMap.delete(resource, context);
+    return DeleteControl.defaultDelete();
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/indexdiscriminator/IndexDiscriminatorTestSpec.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/indexdiscriminator/IndexDiscriminatorTestSpec.java
new file mode 100644
index 0000000000..fcedd48abe
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/indexdiscriminator/IndexDiscriminatorTestSpec.java
@@ -0,0 +1,15 @@
+package io.javaoperatorsdk.operator.sample.indexdiscriminator;
+
+public class IndexDiscriminatorTestSpec {
+
+  private String value;
+
+  public String getValue() {
+    return value;
+  }
+
+  public IndexDiscriminatorTestSpec setValue(String value) {
+    this.value = value;
+    return this;
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/indexdiscriminator/IndexDiscriminatorTestStatus.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/indexdiscriminator/IndexDiscriminatorTestStatus.java
new file mode 100644
index 0000000000..d31c86e8de
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/indexdiscriminator/IndexDiscriminatorTestStatus.java
@@ -0,0 +1,5 @@
+package io.javaoperatorsdk.operator.sample.indexdiscriminator;
+
+public class IndexDiscriminatorTestStatus {
+
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/multipledependentresource/MultipleDependentResourceConfigMap.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/multipledependentresource/MultipleDependentResourceConfigMap.java
index 4cdc2e457d..1adbfb9f95 100644
--- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/multipledependentresource/MultipleDependentResourceConfigMap.java
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/multipledependentresource/MultipleDependentResourceConfigMap.java
@@ -31,7 +31,6 @@ protected ConfigMap desired(MultipleDependentResourceCustomResource primary,
         .withNewMetadata()
         .withName(primary.getConfigMapName(value))
         .withNamespace(primary.getMetadata().getNamespace())
-        .withLabels(Map.of(MultipleDependentResourceReconciler.LABEL, String.valueOf(value)))
         .endMetadata()
         .withData(data)
         .build();
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/multipledependentresource/MultipleDependentResourceReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/multipledependentresource/MultipleDependentResourceReconciler.java
index 8cdbb81eba..49f5ee64c1 100644
--- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/multipledependentresource/MultipleDependentResourceReconciler.java
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/multipledependentresource/MultipleDependentResourceReconciler.java
@@ -3,16 +3,14 @@
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import io.fabric8.kubernetes.api.model.ConfigMap;
 import io.fabric8.kubernetes.client.KubernetesClient;
-import io.javaoperatorsdk.operator.api.reconciler.Context;
-import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
-import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
-import io.javaoperatorsdk.operator.api.reconciler.EventSourceInitializer;
-import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
-import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.api.config.informer.InformerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.*;
 import io.javaoperatorsdk.operator.junit.KubernetesClientAware;
-import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResourceConfig;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
 import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
 import io.javaoperatorsdk.operator.support.TestExecutionInfoProvider;
 
 @ControllerConfiguration
@@ -23,7 +21,6 @@ public class MultipleDependentResourceReconciler
 
   public static final int FIRST_CONFIG_MAP_ID = 1;
   public static final int SECOND_CONFIG_MAP_ID = 2;
-  public static final String LABEL = "multipledependentresource";
   private final AtomicInteger numberOfExecutions = new AtomicInteger(0);
 
   private final MultipleDependentResourceConfigMap firstDependentResourceConfigMap;
@@ -32,18 +29,19 @@ public class MultipleDependentResourceReconciler
 
   public MultipleDependentResourceReconciler() {
     firstDependentResourceConfigMap = new MultipleDependentResourceConfigMap(FIRST_CONFIG_MAP_ID);
-    secondDependentResourceConfigMap = new MultipleDependentResourceConfigMap(SECOND_CONFIG_MAP_ID);
 
-    firstDependentResourceConfigMap.configureWith(
-        new KubernetesDependentResourceConfig()
-            .setLabelSelector(getLabelSelector(FIRST_CONFIG_MAP_ID)));
-    secondDependentResourceConfigMap.configureWith(
-        new KubernetesDependentResourceConfig()
-            .setLabelSelector(getLabelSelector(SECOND_CONFIG_MAP_ID)));
-  }
+    secondDependentResourceConfigMap = new MultipleDependentResourceConfigMap(SECOND_CONFIG_MAP_ID);
 
-  private String getLabelSelector(int resourceId) {
-    return LABEL + "=" + resourceId;
+    firstDependentResourceConfigMap
+        .setResourceDiscriminator(
+            new ResourceIDMatcherDiscriminator<>(
+                p -> new ResourceID(p.getConfigMapName(FIRST_CONFIG_MAP_ID),
+                    p.getMetadata().getNamespace())));
+    secondDependentResourceConfigMap
+        .setResourceDiscriminator(
+            new ResourceIDMatcherDiscriminator<>(
+                p -> new ResourceID(p.getConfigMapName(SECOND_CONFIG_MAP_ID),
+                    p.getMetadata().getNamespace())));
   }
 
   @Override
@@ -64,9 +62,13 @@ public int getNumberOfExecutions() {
   @Override
   public Map<String, EventSource> prepareEventSources(
       EventSourceContext<MultipleDependentResourceCustomResource> context) {
-    return EventSourceInitializer.nameEventSources(
-        firstDependentResourceConfigMap.initEventSource(context),
-        secondDependentResourceConfigMap.initEventSource(context));
+    InformerEventSource<ConfigMap, MultipleDependentResourceCustomResource> eventSource =
+        new InformerEventSource<>(InformerConfiguration.from(ConfigMap.class, context)
+            .build(), context);
+    firstDependentResourceConfigMap.configureWith(eventSource);
+    secondDependentResourceConfigMap.configureWith(eventSource);
+
+    return EventSourceInitializer.nameEventSources(eventSource);
   }
 
   @Override
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/orderedmanageddependent/ConfigMapDependentResource1.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/orderedmanageddependent/ConfigMapDependentResource1.java
index 14530cf17e..bf8d60d9c4 100644
--- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/orderedmanageddependent/ConfigMapDependentResource1.java
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/orderedmanageddependent/ConfigMapDependentResource1.java
@@ -6,11 +6,14 @@
 import io.fabric8.kubernetes.api.model.ConfigMap;
 import io.fabric8.kubernetes.api.model.ObjectMeta;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ResourceIDMatcherDiscriminator;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.ReconcileResult;
 import io.javaoperatorsdk.operator.processing.dependent.kubernetes.CRUDKubernetesDependentResource;
 import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependent;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
 
-@KubernetesDependent(labelSelector = "dependent = cm1")
+@KubernetesDependent(labelSelector = "dependent = cm1",
+    resourceDiscriminator = ConfigMapDependentResource1.CM1ResourceDiscriminator.class)
 public class ConfigMapDependentResource1 extends
     CRUDKubernetesDependentResource<ConfigMap, OrderedManagedDependentCustomResource> {
 
@@ -42,4 +45,11 @@ protected ConfigMap desired(OrderedManagedDependentCustomResource primary,
     return configMap;
   }
 
+  public static class CM1ResourceDiscriminator
+      extends ResourceIDMatcherDiscriminator<ConfigMap, OrderedManagedDependentCustomResource> {
+    public CM1ResourceDiscriminator() {
+      super(p -> new ResourceID(p.getMetadata().getName() + "1", p.getMetadata().getNamespace()));
+    }
+  }
+
 }
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/orderedmanageddependent/ConfigMapDependentResource2.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/orderedmanageddependent/ConfigMapDependentResource2.java
index 35ae69586e..2b17d615b9 100644
--- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/orderedmanageddependent/ConfigMapDependentResource2.java
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/orderedmanageddependent/ConfigMapDependentResource2.java
@@ -6,11 +6,14 @@
 import io.fabric8.kubernetes.api.model.ConfigMap;
 import io.fabric8.kubernetes.api.model.ObjectMeta;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ResourceIDMatcherDiscriminator;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.ReconcileResult;
 import io.javaoperatorsdk.operator.processing.dependent.kubernetes.CRUDKubernetesDependentResource;
 import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependent;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
 
-@KubernetesDependent(labelSelector = "dependent = cm2")
+@KubernetesDependent(labelSelector = "dependent = cm2",
+    resourceDiscriminator = ConfigMapDependentResource2.CM2ResourceDiscriminator.class)
 public class ConfigMapDependentResource2 extends
     CRUDKubernetesDependentResource<ConfigMap, OrderedManagedDependentCustomResource> {
 
@@ -42,4 +45,11 @@ protected ConfigMap desired(OrderedManagedDependentCustomResource primary,
     return configMap;
   }
 
+  public static class CM2ResourceDiscriminator
+      extends ResourceIDMatcherDiscriminator<ConfigMap, OrderedManagedDependentCustomResource> {
+    public CM2ResourceDiscriminator() {
+      super(p -> new ResourceID(p.getMetadata().getName() + "2", p.getMetadata().getNamespace()));
+    }
+  }
+
 }
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/perresourceeventsource/PerResourcePollingEventSourceTestReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/perresourceeventsource/PerResourcePollingEventSourceTestReconciler.java
index dec730d536..ecc33afcb3 100644
--- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/perresourceeventsource/PerResourcePollingEventSourceTestReconciler.java
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/perresourceeventsource/PerResourcePollingEventSourceTestReconciler.java
@@ -57,7 +57,8 @@ public void setKubernetesClient(KubernetesClient kubernetesClient) {
   }
 
   public int getNumberOfExecutions(String name) {
-    return numberOfExecutions.get(name);
+    var num = numberOfExecutions.get(name);
+    return num == null ? 0 : num;
   }
 
   public int getNumberOfFetchExecution(String name) {
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/standalonedependent/StandaloneDependentTestReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/standalonedependent/StandaloneDependentTestReconciler.java
index af6b2d7e25..2dace670ba 100644
--- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/standalonedependent/StandaloneDependentTestReconciler.java
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/standalonedependent/StandaloneDependentTestReconciler.java
@@ -47,7 +47,7 @@ public UpdateControl<StandaloneDependentTestCustomResource> reconcile(
       StandaloneDependentTestCustomResource primary,
       Context<StandaloneDependentTestCustomResource> context) {
     deploymentDependent.reconcile(primary, context);
-    Optional<Deployment> deployment = deploymentDependent.getSecondaryResource(primary);
+    Optional<Deployment> deployment = context.getSecondaryResource(Deployment.class);
     if (deployment.isEmpty()) {
       throw new IllegalStateException("Resource should not be empty after reconcile.");
     }
diff --git a/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageDependentsWorkflowReconciler.java b/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageDependentsWorkflowReconciler.java
index 6986180b89..6b81a9d6e3 100644
--- a/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageDependentsWorkflowReconciler.java
+++ b/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageDependentsWorkflowReconciler.java
@@ -69,7 +69,7 @@ public UpdateControl<WebPage> reconcile(WebPage webPage, Context<WebPage> contex
 
     webPage.setStatus(
         createStatus(
-            configMapDR.getSecondaryResource(webPage).orElseThrow().getMetadata().getName()));
+            context.getSecondaryResource(ConfigMap.class).orElseThrow().getMetadata().getName()));
     return UpdateControl.patchStatus(webPage);
   }
 
@@ -93,6 +93,4 @@ private void initDependentResources(KubernetesClient client) {
     });
   }
 
-
-
 }
diff --git a/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageStandaloneDependentsReconciler.java b/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageStandaloneDependentsReconciler.java
index 0bb692a4d0..b99e130135 100644
--- a/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageStandaloneDependentsReconciler.java
+++ b/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageStandaloneDependentsReconciler.java
@@ -65,7 +65,7 @@ public UpdateControl<WebPage> reconcile(WebPage webPage, Context<WebPage> contex
 
     webPage.setStatus(
         createStatus(
-            configMapDR.getSecondaryResource(webPage).orElseThrow().getMetadata().getName()));
+            context.getSecondaryResource(ConfigMap.class).orElseThrow().getMetadata().getName()));
     return UpdateControl.patchStatus(webPage);
   }