diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/ReconcilerUtils.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/ReconcilerUtils.java
index ea7c58acfb..2a87e045fd 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/ReconcilerUtils.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/ReconcilerUtils.java
@@ -6,14 +6,11 @@
 import java.lang.reflect.Method;
 import java.util.Arrays;
 import java.util.Locale;
-import java.util.Objects;
 import java.util.function.Predicate;
 import java.util.regex.Pattern;
 
 import io.fabric8.kubernetes.api.builder.Builder;
-import io.fabric8.kubernetes.api.model.GenericKubernetesResource;
 import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.fabric8.kubernetes.api.model.Namespaced;
 import io.fabric8.kubernetes.client.CustomResource;
 import io.fabric8.kubernetes.client.KubernetesClientException;
 import io.fabric8.kubernetes.client.utils.Serialization;
@@ -73,36 +70,6 @@ public static String getNameFor(Class<? extends Reconciler> reconcilerClass) {
     return getDefaultNameFor(reconcilerClass);
   }
 
-  public static void checkIfCanAddOwnerReference(HasMetadata owner, HasMetadata resource) {
-    if (owner instanceof GenericKubernetesResource
-        || resource instanceof GenericKubernetesResource) {
-      return;
-    }
-    if (owner instanceof Namespaced) {
-      if (!(resource instanceof Namespaced)) {
-        throw new OperatorException(
-            "Cannot add owner reference from a cluster scoped to a namespace scoped resource."
-                + resourcesIdentifierDescription(owner, resource));
-      } else if (!Objects.equals(
-          owner.getMetadata().getNamespace(), resource.getMetadata().getNamespace())) {
-        throw new OperatorException(
-            "Cannot add owner reference between two resource in different namespaces."
-                + resourcesIdentifierDescription(owner, resource));
-      }
-    }
-  }
-
-  private static String resourcesIdentifierDescription(HasMetadata owner, HasMetadata resource) {
-    return " Owner name: "
-        + owner.getMetadata().getName()
-        + " Kind: "
-        + owner.getKind()
-        + ", Resource name: "
-        + resource.getMetadata().getName()
-        + " Kind: "
-        + resource.getKind();
-  }
-
   public static String getNameFor(Reconciler reconciler) {
     return getNameFor(reconciler.getClass());
   }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/Field.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/Field.java
new file mode 100644
index 0000000000..3acd193cf6
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/Field.java
@@ -0,0 +1,10 @@
+package io.javaoperatorsdk.operator.api.config.informer;
+
+public @interface Field {
+
+  String path();
+
+  String value();
+
+  boolean negated() default false;
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/FieldSelector.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/FieldSelector.java
new file mode 100644
index 0000000000..412ffafdfb
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/FieldSelector.java
@@ -0,0 +1,26 @@
+package io.javaoperatorsdk.operator.api.config.informer;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class FieldSelector {
+  private final List<Field> fields;
+
+  public FieldSelector(List<Field> fields) {
+    this.fields = fields;
+  }
+
+  public FieldSelector(Field... fields) {
+    this.fields = Arrays.asList(fields);
+  }
+
+  public List<Field> getFields() {
+    return fields;
+  }
+
+  public record Field(String path, String value, boolean negated) {
+    public Field(String path, String value) {
+      this(path, value, false);
+    }
+  }
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/FieldSelectorBuilder.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/FieldSelectorBuilder.java
new file mode 100644
index 0000000000..b2cf4d0b5e
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/FieldSelectorBuilder.java
@@ -0,0 +1,23 @@
+package io.javaoperatorsdk.operator.api.config.informer;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class FieldSelectorBuilder {
+
+  private final List<FieldSelector.Field> fields = new ArrayList<>();
+
+  public FieldSelectorBuilder withField(String path, String value) {
+    fields.add(new FieldSelector.Field(path, value));
+    return this;
+  }
+
+  public FieldSelectorBuilder withoutField(String path, String value) {
+    fields.add(new FieldSelector.Field(path, value, true));
+    return this;
+  }
+
+  public FieldSelector build() {
+    return new FieldSelector(fields);
+  }
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/Informer.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/Informer.java
index 80a025009d..cf40da317e 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/Informer.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/Informer.java
@@ -113,4 +113,7 @@
    * the informer cache.
    */
   long informerListLimit() default NO_LONG_VALUE_SET;
+
+  /** Kubernetes field selector for additional resource filtering */
+  Field[] fieldSelector() default {};
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerConfiguration.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerConfiguration.java
index 958a2a7a6f..5fbb62daff 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerConfiguration.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerConfiguration.java
@@ -1,5 +1,6 @@
 package io.javaoperatorsdk.operator.api.config.informer;
 
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Set;
@@ -36,6 +37,7 @@ public class InformerConfiguration<R extends HasMetadata> {
   private GenericFilter<? super R> genericFilter;
   private ItemStore<R> itemStore;
   private Long informerListLimit;
+  private FieldSelector fieldSelector;
 
   protected InformerConfiguration(
       Class<R> resourceClass,
@@ -48,7 +50,8 @@ protected InformerConfiguration(
       OnDeleteFilter<? super R> onDeleteFilter,
       GenericFilter<? super R> genericFilter,
       ItemStore<R> itemStore,
-      Long informerListLimit) {
+      Long informerListLimit,
+      FieldSelector fieldSelector) {
     this(resourceClass);
     this.name = name;
     this.namespaces = namespaces;
@@ -60,6 +63,7 @@ protected InformerConfiguration(
     this.genericFilter = genericFilter;
     this.itemStore = itemStore;
     this.informerListLimit = informerListLimit;
+    this.fieldSelector = fieldSelector;
   }
 
   private InformerConfiguration(Class<R> resourceClass) {
@@ -93,7 +97,8 @@ public static <R extends HasMetadata> InformerConfiguration<R>.Builder builder(
             original.onDeleteFilter,
             original.genericFilter,
             original.itemStore,
-            original.informerListLimit)
+            original.informerListLimit,
+            original.fieldSelector)
         .builder;
   }
 
@@ -264,6 +269,10 @@ public Long getInformerListLimit() {
     return informerListLimit;
   }
 
+  public FieldSelector getFieldSelector() {
+    return fieldSelector;
+  }
+
   @SuppressWarnings("UnusedReturnValue")
   public class Builder {
 
@@ -329,6 +338,12 @@ public InformerConfiguration<R>.Builder initFromAnnotation(
         final var informerListLimit =
             informerListLimitValue == Constants.NO_LONG_VALUE_SET ? null : informerListLimitValue;
         withInformerListLimit(informerListLimit);
+
+        withFieldSelector(
+            new FieldSelector(
+                Arrays.stream(informerConfig.fieldSelector())
+                    .map(f -> new FieldSelector.Field(f.path(), f.value(), f.negated()))
+                    .toList()));
       }
       return this;
     }
@@ -424,5 +439,10 @@ public Builder withInformerListLimit(Long informerListLimit) {
       InformerConfiguration.this.informerListLimit = informerListLimit;
       return this;
     }
+
+    public Builder withFieldSelector(FieldSelector fieldSelector) {
+      InformerConfiguration.this.fieldSelector = fieldSelector;
+      return this;
+    }
   }
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerEventSourceConfiguration.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerEventSourceConfiguration.java
index 2369d5f523..6a38c59bd1 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerEventSourceConfiguration.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerEventSourceConfiguration.java
@@ -265,6 +265,11 @@ public Builder<R> withInformerListLimit(Long informerListLimit) {
       return this;
     }
 
+    public Builder<R> withFieldSelector(FieldSelector fieldSelector) {
+      config.withFieldSelector(fieldSelector);
+      return this;
+    }
+
     public void updateFrom(InformerConfiguration<R> informerConfig) {
       if (informerConfig != null) {
         final var informerConfigName = informerConfig.getName();
@@ -281,7 +286,8 @@ public void updateFrom(InformerConfiguration<R> informerConfig) {
             .withOnUpdateFilter(informerConfig.getOnUpdateFilter())
             .withOnDeleteFilter(informerConfig.getOnDeleteFilter())
             .withGenericFilter(informerConfig.getGenericFilter())
-            .withInformerListLimit(informerConfig.getInformerListLimit());
+            .withInformerListLimit(informerConfig.getInformerListLimit())
+            .withFieldSelector(informerConfig.getFieldSelector());
       }
     }
 
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/BaseControl.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/BaseControl.java
index a5cdb85257..565f2b4d9d 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/BaseControl.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/BaseControl.java
@@ -3,10 +3,17 @@
 import java.time.Duration;
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
+import java.util.function.BiPredicate;
 
-public abstract class BaseControl<T extends BaseControl<T>> {
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.expectation.Expectation;
+import io.javaoperatorsdk.operator.api.reconciler.expectation.ExpectationAdapter;
+import io.javaoperatorsdk.operator.api.reconciler.expectation.ExpectationContext;
+
+public abstract class BaseControl<T extends BaseControl<T, P>, P extends HasMetadata> {
 
   private Long scheduleDelay = null;
+  private Expectation<P> expectation;
 
   public T rescheduleAfter(long delay) {
     rescheduleAfter(Duration.ofMillis(delay));
@@ -25,4 +32,16 @@ public T rescheduleAfter(long delay, TimeUnit timeUnit) {
   public Optional<Long> getScheduleDelay() {
     return Optional.ofNullable(scheduleDelay);
   }
+
+  public void expect(Expectation<P> expectation) {
+    this.expectation = expectation;
+  }
+
+  public void expect(BiPredicate<P, ExpectationContext<P>> expectation, Duration timeout) {
+    this.expectation = new ExpectationAdapter<>(expectation, timeout);
+  }
+
+  public Optional<Expectation<P>> getExpectation() {
+    return Optional.ofNullable(expectation);
+  }
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/CacheAware.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/CacheAware.java
new file mode 100644
index 0000000000..b29732214f
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/CacheAware.java
@@ -0,0 +1,44 @@
+package io.javaoperatorsdk.operator.api.reconciler;
+
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Stream;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
+import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
+import io.javaoperatorsdk.operator.processing.event.source.IndexerResourceCache;
+
+public interface CacheAware<P extends HasMetadata> {
+  default <R> Optional<R> getSecondaryResource(Class<R> expectedType) {
+    return getSecondaryResource(expectedType, null);
+  }
+
+  <R> Set<R> getSecondaryResources(Class<R> expectedType);
+
+  default <R> Stream<R> getSecondaryResourcesAsStream(Class<R> expectedType) {
+    return getSecondaryResources(expectedType).stream();
+  }
+
+  <R> Optional<R> getSecondaryResource(Class<R> expectedType, String eventSourceName);
+
+  ControllerConfiguration<P> getControllerConfiguration();
+
+  /**
+   * Retrieves the primary resource.
+   *
+   * @return the primary resource associated with the current reconciliation
+   */
+  P getPrimaryResource();
+
+  /**
+   * Retrieves the primary resource cache.
+   *
+   * @return the {@link IndexerResourceCache} associated with the associated {@link Reconciler} for
+   *     this context
+   */
+  @SuppressWarnings("unused")
+  IndexedResourceCache<P> getPrimaryCache();
+
+  EventSourceRetriever<P> eventSourceRetriever();
+}
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 f47deb9734..2787bc8ae2 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
@@ -1,35 +1,19 @@
 package io.javaoperatorsdk.operator.api.reconciler;
 
 import java.util.Optional;
-import java.util.Set;
 import java.util.concurrent.ExecutorService;
-import java.util.stream.Stream;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.KubernetesClient;
-import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.ManagedWorkflowAndDependentResourceContext;
-import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
+import io.javaoperatorsdk.operator.api.reconciler.expectation.Expectation;
+import io.javaoperatorsdk.operator.api.reconciler.expectation.ExpectationResult;
 import io.javaoperatorsdk.operator.processing.event.source.IndexerResourceCache;
 
-public interface Context<P extends HasMetadata> {
+public interface Context<P extends HasMetadata> extends CacheAware<P> {
 
   Optional<RetryInfo> getRetryInfo();
 
-  default <R> Optional<R> getSecondaryResource(Class<R> expectedType) {
-    return getSecondaryResource(expectedType, null);
-  }
-
-  <R> Set<R> getSecondaryResources(Class<R> expectedType);
-
-  default <R> Stream<R> getSecondaryResourcesAsStream(Class<R> expectedType) {
-    return getSecondaryResources(expectedType).stream();
-  }
-
-  <R> Optional<R> getSecondaryResource(Class<R> expectedType, String eventSourceName);
-
-  ControllerConfiguration<P> getControllerConfiguration();
-
   /**
    * Retrieve the {@link ManagedWorkflowAndDependentResourceContext} used to interact with {@link
    * io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource}s and associated {@link
@@ -39,8 +23,6 @@ default <R> Stream<R> getSecondaryResourcesAsStream(Class<R> expectedType) {
    */
   ManagedWorkflowAndDependentResourceContext managedWorkflowAndDependentResourceContext();
 
-  EventSourceRetriever<P> eventSourceRetriever();
-
   KubernetesClient getClient();
 
   /** ExecutorService initialized by framework for workflows. Used for workflow standalone mode. */
@@ -72,4 +54,6 @@ default <R> Stream<R> getSecondaryResourcesAsStream(Class<R> expectedType) {
    * @return {@code true} is another reconciliation is already scheduled, {@code false} otherwise
    */
   boolean isNextReconciliationImminent();
+
+  <T extends Expectation<P>> Optional<ExpectationResult<P, T>> expectationResult();
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DefaultCacheAware.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DefaultCacheAware.java
new file mode 100644
index 0000000000..aad14bd860
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DefaultCacheAware.java
@@ -0,0 +1,79 @@
+package io.javaoperatorsdk.operator.api.reconciler;
+
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
+import io.javaoperatorsdk.operator.processing.Controller;
+import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
+import io.javaoperatorsdk.operator.processing.event.NoEventSourceForClassException;
+
+public class DefaultCacheAware<P extends HasMetadata> implements CacheAware<P> {
+
+  protected final Controller<P> controller;
+  protected final P primaryResource;
+
+  public DefaultCacheAware(Controller<P> controller, P primaryResource) {
+    this.controller = controller;
+    this.primaryResource = primaryResource;
+  }
+
+  @Override
+  public <T> Set<T> getSecondaryResources(Class<T> expectedType) {
+    return getSecondaryResourcesAsStream(expectedType).collect(Collectors.toSet());
+  }
+
+  @Override
+  public <R> Stream<R> getSecondaryResourcesAsStream(Class<R> expectedType) {
+    return controller.getEventSourceManager().getEventSourcesFor(expectedType).stream()
+        .map(es -> es.getSecondaryResources(primaryResource))
+        .flatMap(Set::stream);
+  }
+
+  @Override
+  public <T> Optional<T> getSecondaryResource(Class<T> expectedType, String eventSourceName) {
+    try {
+      return controller
+          .getEventSourceManager()
+          .getEventSourceFor(expectedType, eventSourceName)
+          .getSecondaryResource(primaryResource);
+    } catch (NoEventSourceForClassException e) {
+      /*
+       * If a workflow has an activation condition there can be event sources which are only
+       * registered if the activation condition holds, but to provide a consistent API we return an
+       * Optional instead of throwing an exception.
+       *
+       * Note that not only the resource which has an activation condition might not be registered
+       * but dependents which depend on it.
+       */
+      if (eventSourceName == null && controller.workflowContainsDependentForType(expectedType)) {
+        return Optional.empty();
+      } else {
+        throw e;
+      }
+    }
+  }
+
+  @Override
+  public EventSourceRetriever<P> eventSourceRetriever() {
+    return controller.getEventSourceManager();
+  }
+
+  @Override
+  public ControllerConfiguration<P> getControllerConfiguration() {
+    return controller.getConfiguration();
+  }
+
+  @Override
+  public P getPrimaryResource() {
+    return primaryResource;
+  }
+
+  @Override
+  public IndexedResourceCache<P> getPrimaryCache() {
+    return controller.getEventSourceManager().getControllerEventSource();
+  }
+}
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 2acf8d13ca..820e5989ca 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
@@ -1,37 +1,39 @@
 package io.javaoperatorsdk.operator.api.reconciler;
 
 import java.util.Optional;
-import java.util.Set;
 import java.util.concurrent.ExecutorService;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.KubernetesClient;
-import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DefaultManagedWorkflowAndDependentResourceContext;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.ManagedWorkflowAndDependentResourceContext;
+import io.javaoperatorsdk.operator.api.reconciler.expectation.Expectation;
+import io.javaoperatorsdk.operator.api.reconciler.expectation.ExpectationResult;
 import io.javaoperatorsdk.operator.processing.Controller;
-import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
-import io.javaoperatorsdk.operator.processing.event.NoEventSourceForClassException;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 
-public class DefaultContext<P extends HasMetadata> implements Context<P> {
+public class DefaultContext<P extends HasMetadata> extends DefaultCacheAware<P>
+    implements Context<P> {
 
   private RetryInfo retryInfo;
-  private final Controller<P> controller;
-  private final P primaryResource;
-  private final ControllerConfiguration<P> controllerConfiguration;
+
   private final DefaultManagedWorkflowAndDependentResourceContext<P>
       defaultManagedDependentResourceContext;
 
-  public DefaultContext(RetryInfo retryInfo, Controller<P> controller, P primaryResource) {
+  @SuppressWarnings("rawtypes")
+  private final ExpectationResult expectationResult;
+
+  @SuppressWarnings("rawtypes")
+  public DefaultContext(
+      RetryInfo retryInfo,
+      Controller<P> controller,
+      P primaryResource,
+      ExpectationResult expectationResult) {
+    super(controller, primaryResource);
     this.retryInfo = retryInfo;
-    this.controller = controller;
-    this.primaryResource = primaryResource;
-    this.controllerConfiguration = controller.getConfiguration();
     this.defaultManagedDependentResourceContext =
         new DefaultManagedWorkflowAndDependentResourceContext<>(controller, primaryResource, this);
+    this.expectationResult = expectationResult;
   }
 
   @Override
@@ -39,57 +41,11 @@ public Optional<RetryInfo> getRetryInfo() {
     return Optional.ofNullable(retryInfo);
   }
 
-  @Override
-  public <T> Set<T> getSecondaryResources(Class<T> expectedType) {
-    return getSecondaryResourcesAsStream(expectedType).collect(Collectors.toSet());
-  }
-
-  @Override
-  public <R> Stream<R> getSecondaryResourcesAsStream(Class<R> expectedType) {
-    return controller.getEventSourceManager().getEventSourcesFor(expectedType).stream()
-        .map(es -> es.getSecondaryResources(primaryResource))
-        .flatMap(Set::stream);
-  }
-
-  @Override
-  public <T> Optional<T> getSecondaryResource(Class<T> expectedType, String eventSourceName) {
-    try {
-      return controller
-          .getEventSourceManager()
-          .getEventSourceFor(expectedType, eventSourceName)
-          .getSecondaryResource(primaryResource);
-    } catch (NoEventSourceForClassException e) {
-      /*
-       * If a workflow has an activation condition there can be event sources which are only
-       * registered if the activation condition holds, but to provide a consistent API we return an
-       * Optional instead of throwing an exception.
-       *
-       * Note that not only the resource which has an activation condition might not be registered
-       * but dependents which depend on it.
-       */
-      if (eventSourceName == null && controller.workflowContainsDependentForType(expectedType)) {
-        return Optional.empty();
-      } else {
-        throw e;
-      }
-    }
-  }
-
-  @Override
-  public ControllerConfiguration<P> getControllerConfiguration() {
-    return controllerConfiguration;
-  }
-
   @Override
   public ManagedWorkflowAndDependentResourceContext managedWorkflowAndDependentResourceContext() {
     return defaultManagedDependentResourceContext;
   }
 
-  @Override
-  public EventSourceRetriever<P> eventSourceRetriever() {
-    return controller.getEventSourceManager();
-  }
-
   @Override
   public KubernetesClient getClient() {
     return controller.getClient();
@@ -102,16 +58,6 @@ public ExecutorService getWorkflowExecutorService() {
     return controller.getExecutorServiceManager().workflowExecutorService();
   }
 
-  @Override
-  public P getPrimaryResource() {
-    return primaryResource;
-  }
-
-  @Override
-  public IndexedResourceCache<P> getPrimaryCache() {
-    return controller.getEventSourceManager().getControllerEventSource();
-  }
-
   @Override
   public boolean isNextReconciliationImminent() {
     return controller
@@ -119,6 +65,12 @@ public boolean isNextReconciliationImminent() {
         .isNextReconciliationImminent(ResourceID.fromResource(primaryResource));
   }
 
+  @Override
+  @SuppressWarnings("unchecked")
+  public <T extends Expectation<P>> Optional<ExpectationResult<P, T>> expectationResult() {
+    return Optional.ofNullable(expectationResult);
+  }
+
   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/DeleteControl.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DeleteControl.java
index 7160e70830..a7f9104d6d 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DeleteControl.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DeleteControl.java
@@ -1,6 +1,8 @@
 package io.javaoperatorsdk.operator.api.reconciler;
 
-public class DeleteControl extends BaseControl<DeleteControl> {
+import io.fabric8.kubernetes.api.model.HasMetadata;
+
+public class DeleteControl<P extends HasMetadata> extends BaseControl<DeleteControl<P>, P> {
 
   private final boolean removeFinalizer;
 
@@ -27,7 +29,7 @@ public static DeleteControl defaultDelete() {
    *
    * @return delete control that will not remove finalizer.
    */
-  public static DeleteControl noFinalizerRemoval() {
+  public static <T extends HasMetadata> DeleteControl<T> noFinalizerRemoval() {
     return new DeleteControl(false);
   }
 
@@ -36,7 +38,7 @@ public boolean isRemoveFinalizer() {
   }
 
   @Override
-  public DeleteControl rescheduleAfter(long delay) {
+  public DeleteControl<P> rescheduleAfter(long delay) {
     if (removeFinalizer) {
       throw new IllegalStateException("Cannot reschedule cleanup if removing finalizer");
     }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ErrorStatusUpdateControl.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ErrorStatusUpdateControl.java
index e9073d613c..9b21d60883 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ErrorStatusUpdateControl.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ErrorStatusUpdateControl.java
@@ -6,7 +6,7 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 
 public class ErrorStatusUpdateControl<P extends HasMetadata>
-    extends BaseControl<ErrorStatusUpdateControl<P>> {
+    extends BaseControl<ErrorStatusUpdateControl<P>, P> {
 
   private final P resource;
   private boolean noRetry = false;
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/UpdateControl.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/UpdateControl.java
index 1b5eefd7ff..f3ebeb76d9 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/UpdateControl.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/UpdateControl.java
@@ -5,7 +5,7 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.CustomResource;
 
-public class UpdateControl<P extends HasMetadata> extends BaseControl<UpdateControl<P>> {
+public class UpdateControl<P extends HasMetadata> extends BaseControl<UpdateControl<P>, P> {
 
   private final P resource;
   private final boolean patchResource;
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/expectation/AbstractExpectation.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/expectation/AbstractExpectation.java
new file mode 100644
index 0000000000..4821df4236
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/expectation/AbstractExpectation.java
@@ -0,0 +1,22 @@
+package io.javaoperatorsdk.operator.api.reconciler.expectation;
+
+import java.time.Duration;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+
+public abstract class AbstractExpectation<P extends HasMetadata> implements Expectation<P> {
+
+  protected final Duration timeout;
+
+  protected AbstractExpectation(Duration timeout) {
+    this.timeout = timeout;
+  }
+
+  @Override
+  public abstract boolean isFulfilled(P primary, ExpectationContext<P> context);
+
+  @Override
+  public Duration timeout() {
+    return timeout;
+  }
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/expectation/DefaultExpectationContext.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/expectation/DefaultExpectationContext.java
new file mode 100644
index 0000000000..7c08faea48
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/expectation/DefaultExpectationContext.java
@@ -0,0 +1,12 @@
+package io.javaoperatorsdk.operator.api.reconciler.expectation;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.DefaultCacheAware;
+import io.javaoperatorsdk.operator.processing.Controller;
+
+public class DefaultExpectationContext<P extends HasMetadata> extends DefaultCacheAware<P>
+    implements ExpectationContext<P> {
+  public DefaultExpectationContext(Controller<P> controller, P primaryResource) {
+    super(controller, primaryResource);
+  }
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/expectation/Expectation.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/expectation/Expectation.java
new file mode 100644
index 0000000000..47a97bd188
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/expectation/Expectation.java
@@ -0,0 +1,12 @@
+package io.javaoperatorsdk.operator.api.reconciler.expectation;
+
+import java.time.Duration;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+
+public interface Expectation<P extends HasMetadata> {
+
+  boolean isFulfilled(P primary, ExpectationContext<P> context);
+
+  Duration timeout();
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/expectation/ExpectationAdapter.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/expectation/ExpectationAdapter.java
new file mode 100644
index 0000000000..ca0707fc31
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/expectation/ExpectationAdapter.java
@@ -0,0 +1,21 @@
+package io.javaoperatorsdk.operator.api.reconciler.expectation;
+
+import java.time.Duration;
+import java.util.function.BiPredicate;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+
+public class ExpectationAdapter<P extends HasMetadata> extends AbstractExpectation<P> {
+
+  private final BiPredicate<P, ExpectationContext<P>> expectation;
+
+  public ExpectationAdapter(BiPredicate<P, ExpectationContext<P>> expectation, Duration timeout) {
+    super(timeout);
+    this.expectation = expectation;
+  }
+
+  @Override
+  public boolean isFulfilled(P primary, ExpectationContext<P> context) {
+    return expectation.test(primary, context);
+  }
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/expectation/ExpectationContext.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/expectation/ExpectationContext.java
new file mode 100644
index 0000000000..614a58dd41
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/expectation/ExpectationContext.java
@@ -0,0 +1,6 @@
+package io.javaoperatorsdk.operator.api.reconciler.expectation;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.CacheAware;
+
+public interface ExpectationContext<P extends HasMetadata> extends CacheAware<P> {}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/expectation/ExpectationResult.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/expectation/ExpectationResult.java
new file mode 100644
index 0000000000..32ca0a085d
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/expectation/ExpectationResult.java
@@ -0,0 +1,6 @@
+package io.javaoperatorsdk.operator.api.reconciler.expectation;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+
+public record ExpectationResult<P extends HasMetadata, T extends Expectation<P>>(
+    ExpectationStatus status, T expectation) {}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/expectation/ExpectationStatus.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/expectation/ExpectationStatus.java
new file mode 100644
index 0000000000..5ced5093da
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/expectation/ExpectationStatus.java
@@ -0,0 +1,6 @@
+package io.javaoperatorsdk.operator.api.reconciler.expectation;
+
+public enum ExpectationStatus {
+  TIMEOUT,
+  FULFILLED;
+}
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 ebd6089aa7..caf9a305ec 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
@@ -11,7 +11,6 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.api.model.Namespaced;
 import io.fabric8.kubernetes.client.dsl.Resource;
-import io.javaoperatorsdk.operator.ReconcilerUtils;
 import io.javaoperatorsdk.operator.api.config.dependent.Configured;
 import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
@@ -206,7 +205,6 @@ protected Resource<R> prepare(Context<P> context, R desired, P primary, String a
 
   protected void addReferenceHandlingMetadata(R desired, P primary) {
     if (addOwnerReference()) {
-      ReconcilerUtils.checkIfCanAddOwnerReference(primary, desired);
       desired.addOwnerReference(primary);
     } else if (useNonOwnerRefBasedSecondaryToPrimaryMapping()) {
       addSecondaryToPrimaryMapperAnnotations(desired, primary);
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 bdaf575814..9e2b3f0416 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
@@ -17,6 +17,10 @@
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.monitoring.Metrics;
 import io.javaoperatorsdk.operator.api.reconciler.Constants;
+import io.javaoperatorsdk.operator.api.reconciler.expectation.DefaultExpectationContext;
+import io.javaoperatorsdk.operator.api.reconciler.expectation.ExpectationContext;
+import io.javaoperatorsdk.operator.api.reconciler.expectation.ExpectationResult;
+import io.javaoperatorsdk.operator.api.reconciler.expectation.ExpectationStatus;
 import io.javaoperatorsdk.operator.processing.LifecycleAware;
 import io.javaoperatorsdk.operator.processing.MDCUtils;
 import io.javaoperatorsdk.operator.processing.event.rate.RateLimiter;
@@ -129,6 +133,7 @@ private void handleMarkedEventForResource(ResourceState state) {
     }
   }
 
+  @SuppressWarnings("rawtypes")
   private void submitReconciliationExecution(ResourceState state) {
     try {
       boolean controllerUnderExecution = isControllerUnderExecution(state);
@@ -136,6 +141,20 @@ private void submitReconciliationExecution(ResourceState state) {
       Optional<P> maybeLatest = cache.get(resourceID);
       maybeLatest.ifPresent(MDCUtils::addResourceInfo);
       if (!controllerUnderExecution && maybeLatest.isPresent()) {
+        ExpectationResult expectationResult = null;
+        if (isExpectationPresent(state)) {
+          var expectationCheckResult =
+              shouldProceedWithExpectation(state, maybeLatest.orElseThrow());
+          if (expectationCheckResult.isEmpty()) {
+            log.debug(
+                "Skipping processing since expectation is not fulfilled. ResourceID: {}",
+                resourceID);
+            return;
+          } else {
+            expectationResult = expectationCheckResult.orElseThrow();
+          }
+        }
+
         var rateLimit = state.getRateLimit();
         if (rateLimit == null) {
           rateLimit = rateLimiter.initState();
@@ -148,7 +167,8 @@ private void submitReconciliationExecution(ResourceState state) {
         }
         state.setUnderProcessing(true);
         final var latest = maybeLatest.get();
-        ExecutionScope<P> executionScope = new ExecutionScope<>(state.getRetry());
+        ExecutionScope<P> executionScope =
+            new ExecutionScope<>(state.getRetry(), expectationResult);
         state.unMarkEventReceived();
         metrics.reconcileCustomResource(latest, state.getRetry(), metricsMetadata);
         log.debug("Executing events for custom resource. Scope: {}", executionScope);
@@ -174,6 +194,24 @@ private void submitReconciliationExecution(ResourceState state) {
     }
   }
 
+  private boolean isExpectationPresent(ResourceState state) {
+    return state.getExpectationHolder().isPresent();
+  }
+
+  @SuppressWarnings({"unchecked", "rawtypes"})
+  Optional<ExpectationResult> shouldProceedWithExpectation(ResourceState state, P primary) {
+
+    var holder = state.getExpectationHolder().orElseThrow();
+    if (holder.isTimedOut()) {
+      return Optional.of(new ExpectationResult(ExpectationStatus.TIMEOUT, holder.getExpectation()));
+    }
+    ExpectationContext<P> expectationContext =
+        new DefaultExpectationContext<>(this.eventSourceManager.getController(), primary);
+    return holder.getExpectation().isFulfilled(primary, expectationContext)
+        ? Optional.of(new ExpectationResult(ExpectationStatus.FULFILLED, holder.getExpectation()))
+        : Optional.empty();
+  }
+
   private void handleEventMarking(Event event, ResourceState state) {
     final var relatedCustomResourceID = event.getRelatedCustomResourceID();
     if (event instanceof ResourceEvent resourceEvent) {
@@ -257,6 +295,9 @@ synchronized void eventProcessingFinished(
       state.markProcessedMarkForDeletion();
       metrics.cleanupDoneFor(resourceID, metricsMetadata);
     } else {
+      // TODO what should be the relation between re-schedule and expectation
+      // should we add a flag if trigger if expectation fails
+      setExpectation(state, postExecutionControl);
       if (state.eventPresent()) {
         submitReconciliationExecution(state);
       } else {
@@ -265,6 +306,10 @@ synchronized void eventProcessingFinished(
     }
   }
 
+  private void setExpectation(ResourceState state, PostExecutionControl<P> postExecutionControl) {
+    postExecutionControl.getExpectation().ifPresent(state::setExpectation);
+  }
+
   /**
    * In case retry is configured more complex error logging takes place, see handleRetryOnException
    */
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ExecutionScope.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ExecutionScope.java
index 90899a6e1a..e42340dbbe 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ExecutionScope.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ExecutionScope.java
@@ -2,15 +2,18 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
+import io.javaoperatorsdk.operator.api.reconciler.expectation.ExpectationResult;
 
 class ExecutionScope<R extends HasMetadata> {
 
   // the latest custom resource from cache
   private R resource;
   private final RetryInfo retryInfo;
+  private final ExpectationResult<R, ?> expectationResult;
 
-  ExecutionScope(RetryInfo retryInfo) {
+  ExecutionScope(RetryInfo retryInfo, ExpectationResult<R, ?> expectationResult) {
     this.retryInfo = retryInfo;
+    this.expectationResult = expectationResult;
   }
 
   public ExecutionScope<R> setResource(R resource) {
@@ -42,4 +45,8 @@ public String toString() {
   public RetryInfo getRetryInfo() {
     return retryInfo;
   }
+
+  public ExpectationResult<R, ?> getExpectationResult() {
+    return expectationResult;
+  }
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ExpectationHolder.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ExpectationHolder.java
new file mode 100644
index 0000000000..a50ab32615
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ExpectationHolder.java
@@ -0,0 +1,37 @@
+package io.javaoperatorsdk.operator.processing.event;
+
+import java.time.LocalDateTime;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.expectation.Expectation;
+
+public class ExpectationHolder<P extends HasMetadata> {
+
+  private LocalDateTime expectationCreationTime;
+  private Expectation<P> expectation;
+
+  public ExpectationHolder(LocalDateTime expectationCreationTime, Expectation<P> expectation) {
+    this.expectationCreationTime = expectationCreationTime;
+    this.expectation = expectation;
+  }
+
+  public LocalDateTime getExpectationCreationTime() {
+    return expectationCreationTime;
+  }
+
+  public void setExpectationCreationTime(LocalDateTime expectationCreationTime) {
+    this.expectationCreationTime = expectationCreationTime;
+  }
+
+  public Expectation<?> getExpectation() {
+    return expectation;
+  }
+
+  public void setExpectation(Expectation<P> expectation) {
+    this.expectation = expectation;
+  }
+
+  public boolean isTimedOut() {
+    return expectationCreationTime.plus(expectation.timeout()).isBefore(LocalDateTime.now());
+  }
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/PostExecutionControl.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/PostExecutionControl.java
index 42311c1cb5..3841a80960 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/PostExecutionControl.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/PostExecutionControl.java
@@ -3,6 +3,7 @@
 import java.util.Optional;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.expectation.Expectation;
 
 final class PostExecutionControl<R extends HasMetadata> {
 
@@ -10,6 +11,7 @@ final class PostExecutionControl<R extends HasMetadata> {
   private final R updatedCustomResource;
   private final boolean updateIsStatusPatch;
   private final Exception runtimeException;
+  private Expectation<R> expectation;
 
   private Long reScheduleDelay = null;
 
@@ -66,6 +68,11 @@ public PostExecutionControl<R> withReSchedule(long delay) {
     return this;
   }
 
+  public PostExecutionControl<R> withExpectation(Expectation<R> expectation) {
+    this.expectation = expectation;
+    return this;
+  }
+
   public Optional<Exception> getRuntimeException() {
     return Optional.ofNullable(runtimeException);
   }
@@ -93,4 +100,8 @@ public String toString() {
   public boolean isFinalizerRemoved() {
     return finalizerRemoved;
   }
+
+  public Optional<Expectation<R>> getExpectation() {
+    return Optional.ofNullable(expectation);
+  }
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcher.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcher.java
index c4b161ef27..c944cb5f3a 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcher.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcher.java
@@ -90,7 +90,11 @@ private PostExecutionControl<P> handleDispatch(ExecutionScope<P> executionScope)
     }
 
     Context<P> context =
-        new DefaultContext<>(executionScope.getRetryInfo(), controller, resourceForExecution);
+        new DefaultContext<>(
+            executionScope.getRetryInfo(),
+            controller,
+            resourceForExecution,
+            executionScope.getExpectationResult());
     if (markedForDeletion) {
       return handleCleanup(resourceForExecution, originalResource, context);
     } else {
@@ -246,11 +250,18 @@ private PostExecutionControl<P> createPostExecutionControl(
       postExecutionControl = PostExecutionControl.defaultDispatch();
     }
     updatePostExecutionControlWithReschedule(postExecutionControl, updateControl);
+    updatePostExecutionControlWithExpectation(postExecutionControl, updateControl);
+
     return postExecutionControl;
   }
 
+  private void updatePostExecutionControlWithExpectation(
+      PostExecutionControl<P> postExecutionControl, BaseControl<?, P> baseControl) {
+    baseControl.getExpectation().ifPresent(postExecutionControl::withExpectation);
+  }
+
   private void updatePostExecutionControlWithReschedule(
-      PostExecutionControl<P> postExecutionControl, BaseControl<?> baseControl) {
+      PostExecutionControl<P> postExecutionControl, BaseControl<?, ?> baseControl) {
     baseControl.getScheduleDelay().ifPresent(postExecutionControl::withReSchedule);
   }
 
@@ -262,7 +273,7 @@ private PostExecutionControl<P> handleCleanup(
           ResourceID.fromResource(resourceForExecution),
           getVersion(resourceForExecution));
     }
-    DeleteControl deleteControl = controller.cleanup(resourceForExecution, context);
+    DeleteControl<P> deleteControl = controller.cleanup(resourceForExecution, context);
     final var useFinalizer = controller.useFinalizer();
     if (useFinalizer) {
       // note that we don't reschedule here even if instructed. Removing finalizer means that
@@ -299,6 +310,7 @@ private PostExecutionControl<P> handleCleanup(
         useFinalizer);
     PostExecutionControl<P> postExecutionControl = PostExecutionControl.defaultDispatch();
     updatePostExecutionControlWithReschedule(postExecutionControl, deleteControl);
+    updatePostExecutionControlWithExpectation(postExecutionControl, deleteControl);
     return postExecutionControl;
   }
 
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ResourceState.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ResourceState.java
index 5d4e74d681..bad045b261 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ResourceState.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ResourceState.java
@@ -1,5 +1,9 @@
 package io.javaoperatorsdk.operator.processing.event;
 
+import java.time.LocalDateTime;
+import java.util.Optional;
+
+import io.javaoperatorsdk.operator.api.reconciler.expectation.Expectation;
 import io.javaoperatorsdk.operator.processing.event.rate.RateLimiter.RateLimitState;
 import io.javaoperatorsdk.operator.processing.retry.RetryExecution;
 
@@ -29,6 +33,7 @@ private enum EventingState {
   private RetryExecution retry;
   private EventingState eventing;
   private RateLimitState rateLimit;
+  private ExpectationHolder expectationHolder;
 
   public ResourceState(ResourceID id) {
     this.id = id;
@@ -75,6 +80,18 @@ public boolean processedMarkForDeletionPresent() {
     return eventing == EventingState.PROCESSED_MARK_FOR_DELETION;
   }
 
+  public void setExpectation(Expectation<?> expectation) {
+    expectationHolder = new ExpectationHolder(LocalDateTime.now(), expectation);
+  }
+
+  public void cleanExpectation() {
+    expectationHolder = null;
+  }
+
+  public Optional<ExpectationHolder> getExpectationHolder() {
+    return Optional.ofNullable(expectationHolder);
+  }
+
   public void markEventReceived() {
     if (deleteEventPresent()) {
       throw new IllegalStateException("Cannot receive event after a delete event received");
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManager.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManager.java
index 1e1607dd8b..f833edffe6 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManager.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManager.java
@@ -134,6 +134,18 @@ private InformerWrapper<R> createEventSource(
       ResourceEventHandler<R> eventHandler,
       String namespaceIdentifier) {
     final var informerConfig = configuration.getInformerConfig();
+
+    if (informerConfig.getFieldSelector() != null
+        && !informerConfig.getFieldSelector().getFields().isEmpty()) {
+      for (var f : informerConfig.getFieldSelector().getFields()) {
+        if (f.negated()) {
+          filteredBySelectorClient = filteredBySelectorClient.withoutField(f.path(), f.value());
+        } else {
+          filteredBySelectorClient = filteredBySelectorClient.withField(f.path(), f.value());
+        }
+      }
+    }
+
     var informer =
         Optional.ofNullable(informerConfig.getInformerListLimit())
             .map(filteredBySelectorClient::withLimit)
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/ReconcilerUtilsTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/ReconcilerUtilsTest.java
index abc83b94ff..6445373c78 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/ReconcilerUtilsTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/ReconcilerUtilsTest.java
@@ -8,8 +8,6 @@
 import io.fabric8.kubernetes.api.model.apps.Deployment;
 import io.fabric8.kubernetes.api.model.apps.DeploymentBuilder;
 import io.fabric8.kubernetes.api.model.apps.DeploymentSpec;
-import io.fabric8.kubernetes.api.model.rbac.ClusterRole;
-import io.fabric8.kubernetes.api.model.rbac.ClusterRoleBuilder;
 import io.fabric8.kubernetes.client.CustomResource;
 import io.fabric8.kubernetes.client.KubernetesClientException;
 import io.fabric8.kubernetes.client.http.HttpRequest;
@@ -154,44 +152,6 @@ void handleKubernetesExceptionShouldThrowMissingCRDExceptionWhenAppropriate() {
                 HasMetadata.getFullResourceName(Tomcat.class)));
   }
 
-  @Test
-  void checksIfOwnerReferenceCanBeAdded() {
-    assertThrows(
-        OperatorException.class,
-        () ->
-            ReconcilerUtils.checkIfCanAddOwnerReference(
-                namespacedResource(), namespacedResourceFromOtherNamespace()));
-
-    assertThrows(
-        OperatorException.class,
-        () ->
-            ReconcilerUtils.checkIfCanAddOwnerReference(
-                namespacedResource(), clusterScopedResource()));
-
-    assertDoesNotThrow(
-        () -> {
-          ReconcilerUtils.checkIfCanAddOwnerReference(
-              clusterScopedResource(), clusterScopedResource());
-          ReconcilerUtils.checkIfCanAddOwnerReference(namespacedResource(), namespacedResource());
-        });
-  }
-
-  private ClusterRole clusterScopedResource() {
-    return new ClusterRoleBuilder().withMetadata(new ObjectMetaBuilder().build()).build();
-  }
-
-  private ConfigMap namespacedResource() {
-    return new ConfigMapBuilder()
-        .withMetadata(new ObjectMetaBuilder().withNamespace("testns1").build())
-        .build();
-  }
-
-  private ConfigMap namespacedResourceFromOtherNamespace() {
-    return new ConfigMapBuilder()
-        .withMetadata(new ObjectMetaBuilder().withNamespace("testns2").build())
-        .build();
-  }
-
   @Group("tomcatoperator.io")
   @Version("v1")
   @ShortNames("tc")
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContextTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContextTest.java
index b289d68b22..de3e1b4221 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContextTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContextTest.java
@@ -18,7 +18,8 @@ class DefaultContextTest {
   private final Secret primary = new Secret();
   private final Controller<Secret> mockController = mock();
 
-  private final DefaultContext<?> context = new DefaultContext<>(null, mockController, primary);
+  private final DefaultContext<?> context =
+      new DefaultContext<>(null, mockController, primary, null);
 
   @Test
   @SuppressWarnings("unchecked")
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/ControllerTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/ControllerTest.java
index 82ecdb111a..8b57506e26 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/ControllerTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/ControllerTest.java
@@ -122,7 +122,7 @@ void callsCleanupOnWorkflowWhenHasCleanerAndReconcilerIsNotCleaner(
         new Controller<Secret>(
             reconciler, configuration, MockKubernetesClient.client(Secret.class));
 
-    controller.cleanup(new Secret(), new DefaultContext<>(null, controller, new Secret()));
+    controller.cleanup(new Secret(), new DefaultContext<>(null, controller, new Secret(), null));
 
     verify(managedWorkflowMock, times(workflowCleanerExecuted ? 1 : 0)).cleanup(any(), any());
   }
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventProcessorTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventProcessorTest.java
index fe2e6e9514..0f45ad77f7 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventProcessorTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventProcessorTest.java
@@ -129,7 +129,7 @@ void ifExecutionInProgressWaitsUntilItsFinished() {
   void schedulesAnEventRetryOnException() {
     TestCustomResource customResource = testCustomResource();
 
-    ExecutionScope executionScope = new ExecutionScope(null);
+    ExecutionScope executionScope = new ExecutionScope(null, null);
     executionScope.setResource(customResource);
     PostExecutionControl postExecutionControl =
         PostExecutionControl.exceptionDuringExecution(new RuntimeException("test"));
@@ -271,7 +271,7 @@ void cancelScheduleOnceEventsOnSuccessfulExecution() {
     var cr = testCustomResource(crID);
 
     eventProcessor.eventProcessingFinished(
-        new ExecutionScope(null).setResource(cr), PostExecutionControl.defaultDispatch());
+        new ExecutionScope(null, null).setResource(cr), PostExecutionControl.defaultDispatch());
 
     verify(retryTimerEventSourceMock, times(1)).cancelOnceSchedule(eq(crID));
   }
@@ -300,7 +300,7 @@ void startProcessedMarkedEventReceivedBefore() {
   @Test
   void notUpdatesEventSourceHandlerIfResourceUpdated() {
     TestCustomResource customResource = testCustomResource();
-    ExecutionScope executionScope = new ExecutionScope(null).setResource(customResource);
+    ExecutionScope executionScope = new ExecutionScope(null, null).setResource(customResource);
     PostExecutionControl postExecutionControl =
         PostExecutionControl.customResourceStatusPatched(customResource);
 
@@ -313,7 +313,7 @@ void notUpdatesEventSourceHandlerIfResourceUpdated() {
   void notReschedulesAfterTheFinalizerRemoveProcessed() {
     TestCustomResource customResource = testCustomResource();
     markForDeletion(customResource);
-    ExecutionScope executionScope = new ExecutionScope(null).setResource(customResource);
+    ExecutionScope executionScope = new ExecutionScope(null, null).setResource(customResource);
     PostExecutionControl postExecutionControl =
         PostExecutionControl.customResourceFinalizerRemoved(customResource);
 
@@ -326,7 +326,7 @@ void notReschedulesAfterTheFinalizerRemoveProcessed() {
   void skipEventProcessingIfFinalizerRemoveProcessed() {
     TestCustomResource customResource = testCustomResource();
     markForDeletion(customResource);
-    ExecutionScope executionScope = new ExecutionScope(null).setResource(customResource);
+    ExecutionScope executionScope = new ExecutionScope(null, null).setResource(customResource);
     PostExecutionControl postExecutionControl =
         PostExecutionControl.customResourceFinalizerRemoved(customResource);
 
@@ -343,7 +343,7 @@ void skipEventProcessingIfFinalizerRemoveProcessed() {
   void newResourceAfterMissedDeleteEvent() {
     TestCustomResource customResource = testCustomResource();
     markForDeletion(customResource);
-    ExecutionScope executionScope = new ExecutionScope(null).setResource(customResource);
+    ExecutionScope executionScope = new ExecutionScope(null, null).setResource(customResource);
     PostExecutionControl postExecutionControl =
         PostExecutionControl.customResourceFinalizerRemoved(customResource);
     var newResource = testCustomResource();
@@ -379,7 +379,7 @@ void rateLimitsReconciliationSubmission() {
   @Test
   void schedulesRetryForMarReconciliationInterval() {
     TestCustomResource customResource = testCustomResource();
-    ExecutionScope executionScope = new ExecutionScope(null).setResource(customResource);
+    ExecutionScope executionScope = new ExecutionScope(null, null).setResource(customResource);
     PostExecutionControl postExecutionControl = PostExecutionControl.defaultDispatch();
 
     eventProcessorWithRetry.eventProcessingFinished(executionScope, postExecutionControl);
@@ -401,7 +401,8 @@ void schedulesRetryForMarReconciliationIntervalIfRetryExhausted() {
                 eventSourceManagerMock,
                 metricsMock));
     eventProcessorWithRetry.start();
-    ExecutionScope executionScope = new ExecutionScope(null).setResource(testCustomResource());
+    ExecutionScope executionScope =
+        new ExecutionScope(null, null).setResource(testCustomResource());
     PostExecutionControl postExecutionControl =
         PostExecutionControl.exceptionDuringExecution(new RuntimeException());
     when(eventProcessorWithRetry.retryEventSource()).thenReturn(retryTimerEventSourceMock);
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcherTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcherTest.java
index 89f3655356..5e4550474b 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcherTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcherTest.java
@@ -404,7 +404,8 @@ public int getAttemptCount() {
                   public boolean isLastAttempt() {
                     return true;
                   }
-                })
+                },
+                null)
             .setResource(testCustomResource));
 
     ArgumentCaptor<Context> contextArgumentCaptor = ArgumentCaptor.forClass(Context.class);
@@ -505,7 +506,8 @@ public int getAttemptCount() {
                   public boolean isLastAttempt() {
                     return true;
                   }
-                })
+                },
+                null)
             .setResource(testCustomResource));
 
     verify(customResourceFacade, times(1)).patchStatus(eq(testCustomResource), any());
@@ -528,7 +530,7 @@ void callErrorStatusHandlerEvenOnFirstError() {
 
     var postExecControl =
         reconciliationDispatcher.handleExecution(
-            new ExecutionScope(null).setResource(testCustomResource));
+            new ExecutionScope(null, null).setResource(testCustomResource));
     verify(customResourceFacade, times(1)).patchStatus(eq(testCustomResource), any());
     verify(reconciler, times(1)).updateErrorStatus(eq(testCustomResource), any(), any());
     assertThat(postExecControl.exceptionDuringExecution()).isTrue();
@@ -549,7 +551,7 @@ void errorHandlerCanInstructNoRetryWithUpdate() {
 
     var postExecControl =
         reconciliationDispatcher.handleExecution(
-            new ExecutionScope(null).setResource(testCustomResource));
+            new ExecutionScope(null, null).setResource(testCustomResource));
 
     verify(reconciler, times(1)).updateErrorStatus(eq(testCustomResource), any(), any());
     verify(customResourceFacade, times(1)).patchStatus(eq(testCustomResource), any());
@@ -571,7 +573,7 @@ void errorHandlerCanInstructNoRetryNoUpdate() {
 
     var postExecControl =
         reconciliationDispatcher.handleExecution(
-            new ExecutionScope(null).setResource(testCustomResource));
+            new ExecutionScope(null, null).setResource(testCustomResource));
 
     verify(reconciler, times(1)).updateErrorStatus(eq(testCustomResource), any(), any());
     verify(customResourceFacade, times(0)).patchStatus(eq(testCustomResource), any());
@@ -588,7 +590,7 @@ void errorStatusHandlerCanPatchResource() {
     reconciler.errorHandler = () -> ErrorStatusUpdateControl.patchStatus(testCustomResource);
 
     reconciliationDispatcher.handleExecution(
-        new ExecutionScope(null).setResource(testCustomResource));
+        new ExecutionScope(null, null).setResource(testCustomResource));
 
     verify(customResourceFacade, times(1)).patchStatus(eq(testCustomResource), any());
     verify(reconciler, times(1)).updateErrorStatus(eq(testCustomResource), any(), any());
@@ -611,7 +613,7 @@ void ifRetryLimitedToZeroMaxAttemptsErrorHandlerGetsCorrectLastAttempt() {
     reconciler.errorHandler = () -> ErrorStatusUpdateControl.noStatusUpdate();
 
     reconciliationDispatcher.handleExecution(
-        new ExecutionScope(null).setResource(testCustomResource));
+        new ExecutionScope(null, null).setResource(testCustomResource));
 
     verify(reconciler, times(1))
         .updateErrorStatus(
@@ -675,7 +677,7 @@ void reSchedulesFromErrorHandler() {
 
     var res =
         reconciliationDispatcher.handleExecution(
-            new ExecutionScope(null).setResource(testCustomResource));
+            new ExecutionScope(null, null).setResource(testCustomResource));
 
     assertThat(res.getReScheduleDelay()).contains(delay);
     assertThat(res.getRuntimeException()).isEmpty();
@@ -723,7 +725,7 @@ private void removeFinalizers(CustomResource customResource) {
   }
 
   public <T extends HasMetadata> ExecutionScope<T> executionScopeWithCREvent(T resource) {
-    return (ExecutionScope<T>) new ExecutionScope<>(null).setResource(resource);
+    return (ExecutionScope<T>) new ExecutionScope<>(null, null).setResource(resource);
   }
 
   private class TestReconciler
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/simple/SimpleExpectationCustomResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/simple/SimpleExpectationCustomResource.java
new file mode 100644
index 0000000000..abec4d3793
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/simple/SimpleExpectationCustomResource.java
@@ -0,0 +1,15 @@
+package io.javaoperatorsdk.operator.baseapi.expectation.simple;
+
+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("secr")
+public class SimpleExpectationCustomResource
+    extends CustomResource<
+        SimpleExpectationCustomResourceSpec, SimpleExpectationCustomResourceStatus>
+    implements Namespaced {}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/simple/SimpleExpectationCustomResourceSpec.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/simple/SimpleExpectationCustomResourceSpec.java
new file mode 100644
index 0000000000..9e33c9fe47
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/simple/SimpleExpectationCustomResourceSpec.java
@@ -0,0 +1,14 @@
+package io.javaoperatorsdk.operator.baseapi.expectation.simple;
+
+public class SimpleExpectationCustomResourceSpec {
+
+  private int replicaCount;
+
+  public int getReplicaCount() {
+    return replicaCount;
+  }
+
+  public void setReplicaCount(int replicaCount) {
+    this.replicaCount = replicaCount;
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/simple/SimpleExpectationCustomResourceStatus.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/simple/SimpleExpectationCustomResourceStatus.java
new file mode 100644
index 0000000000..879b267a33
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/simple/SimpleExpectationCustomResourceStatus.java
@@ -0,0 +1,14 @@
+package io.javaoperatorsdk.operator.baseapi.expectation.simple;
+
+public class SimpleExpectationCustomResourceStatus {
+
+  private Boolean ready;
+
+  public Boolean getReady() {
+    return ready;
+  }
+
+  public void setReady(Boolean ready) {
+    this.ready = ready;
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/simple/SimpleExpectationReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/simple/SimpleExpectationReconciler.java
new file mode 100644
index 0000000000..ae2ffad682
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/simple/SimpleExpectationReconciler.java
@@ -0,0 +1,17 @@
+package io.javaoperatorsdk.operator.baseapi.expectation.simple;
+
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+
+@ControllerConfiguration
+public class SimpleExpectationReconciler implements Reconciler<SimpleExpectationCustomResource> {
+
+  @Override
+  public UpdateControl<SimpleExpectationCustomResource> reconcile(
+      SimpleExpectationCustomResource resource, Context<SimpleExpectationCustomResource> context) {
+
+    return UpdateControl.noUpdate();
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/simple/SimpleExpectationSampleIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/simple/SimpleExpectationSampleIT.java
new file mode 100644
index 0000000000..c177edfbb4
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/simple/SimpleExpectationSampleIT.java
@@ -0,0 +1,29 @@
+package io.javaoperatorsdk.operator.baseapi.expectation.simple;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension;
+
+class SimpleExpectationSampleIT {
+
+  @RegisterExtension
+  LocallyRunOperatorExtension operator =
+      LocallyRunOperatorExtension.builder()
+          .withReconciler(new SimpleExpectationReconciler())
+          .build();
+
+  @Test
+  void exceptDeploymentUp() {}
+
+  public SimpleExpectationCustomResource createCustomResource() {
+    SimpleExpectationCustomResource resource = new SimpleExpectationCustomResource();
+    resource.setMetadata(
+        new ObjectMetaBuilder()
+            .withName("error-status-test")
+            .withNamespace(operator.getNamespace())
+            .build());
+    return resource;
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/fieldselector/FieldSelectorIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/fieldselector/FieldSelectorIT.java
new file mode 100644
index 0000000000..5b32f15265
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/fieldselector/FieldSelectorIT.java
@@ -0,0 +1,73 @@
+package io.javaoperatorsdk.operator.baseapi.fieldselector;
+
+import java.time.Duration;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.fabric8.kubernetes.api.model.SecretBuilder;
+import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+
+import static io.javaoperatorsdk.operator.baseapi.fieldselector.FieldSelectorTestReconciler.MY_SECRET_TYPE;
+import static io.javaoperatorsdk.operator.baseapi.fieldselector.FieldSelectorTestReconciler.OTHER_SECRET_TYPE;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+class FieldSelectorIT {
+
+  public static final String TEST_1 = "test1";
+  public static final String TEST_2 = "test2";
+  public static final String TEST_3 = "test3";
+
+  @RegisterExtension
+  LocallyRunOperatorExtension extension =
+      LocallyRunOperatorExtension.builder()
+          .withReconciler(new FieldSelectorTestReconciler())
+          .build();
+
+  @Test
+  void filtersCustomResourceByLabel() {
+
+    var customPrimarySecret =
+        extension.create(
+            new SecretBuilder()
+                .withMetadata(new ObjectMetaBuilder().withName(TEST_1).build())
+                .withType(MY_SECRET_TYPE)
+                .build());
+
+    var otherSecret =
+        extension.create(
+            new SecretBuilder()
+                .withMetadata(new ObjectMetaBuilder().withName(TEST_2).build())
+                .build());
+
+    var dependentSecret =
+        extension.create(
+            new SecretBuilder()
+                .withMetadata(new ObjectMetaBuilder().withName(TEST_3).build())
+                .withType(OTHER_SECRET_TYPE)
+                .build());
+
+    await()
+        .pollDelay(Duration.ofMillis(150))
+        .untilAsserted(
+            () -> {
+              var r = extension.getReconcilerOfType(FieldSelectorTestReconciler.class);
+              assertThat(r.getReconciledSecrets()).containsExactly(TEST_1);
+
+              assertThat(
+                      r.getDependentSecretEventSource()
+                          .get(ResourceID.fromResource(dependentSecret)))
+                  .isPresent();
+              assertThat(
+                      r.getDependentSecretEventSource()
+                          .get(ResourceID.fromResource(customPrimarySecret)))
+                  .isNotPresent();
+              assertThat(
+                      r.getDependentSecretEventSource().get(ResourceID.fromResource(otherSecret)))
+                  .isNotPresent();
+            });
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/fieldselector/FieldSelectorTestReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/fieldselector/FieldSelectorTestReconciler.java
new file mode 100644
index 0000000000..1e3fddcf83
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/fieldselector/FieldSelectorTestReconciler.java
@@ -0,0 +1,69 @@
+package io.javaoperatorsdk.operator.baseapi.fieldselector;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.fabric8.kubernetes.api.model.Secret;
+import io.javaoperatorsdk.operator.api.config.informer.Field;
+import io.javaoperatorsdk.operator.api.config.informer.FieldSelectorBuilder;
+import io.javaoperatorsdk.operator.api.config.informer.Informer;
+import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration;
+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.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
+import io.javaoperatorsdk.operator.support.TestExecutionInfoProvider;
+
+@ControllerConfiguration(
+    informer =
+        @Informer(
+            fieldSelector =
+                @Field(path = "type", value = FieldSelectorTestReconciler.MY_SECRET_TYPE)))
+public class FieldSelectorTestReconciler implements Reconciler<Secret>, TestExecutionInfoProvider {
+
+  public static final String MY_SECRET_TYPE = "my-secret-type";
+  public static final String OTHER_SECRET_TYPE = "my-dependent-secret-type";
+  private final AtomicInteger numberOfExecutions = new AtomicInteger(0);
+
+  private Set<String> reconciledSecrets = Collections.synchronizedSet(new HashSet<>());
+  private InformerEventSource<Secret, Secret> dependentSecretEventSource;
+
+  @Override
+  public UpdateControl<Secret> reconcile(Secret resource, Context<Secret> context) {
+    reconciledSecrets.add(resource.getMetadata().getName());
+    numberOfExecutions.addAndGet(1);
+    return UpdateControl.noUpdate();
+  }
+
+  public int getNumberOfExecutions() {
+    return numberOfExecutions.get();
+  }
+
+  public Set<String> getReconciledSecrets() {
+    return reconciledSecrets;
+  }
+
+  @Override
+  public List<EventSource<?, Secret>> prepareEventSources(EventSourceContext<Secret> context) {
+    dependentSecretEventSource =
+        new InformerEventSource<>(
+            InformerEventSourceConfiguration.from(Secret.class, Secret.class)
+                .withNamespacesInheritedFromController()
+                .withFieldSelector(
+                    new FieldSelectorBuilder().withField("type", OTHER_SECRET_TYPE).build())
+                .build(),
+            context);
+
+    return List.of(dependentSecretEventSource);
+  }
+
+  public InformerEventSource<Secret, Secret> getDependentSecretEventSource() {
+    return dependentSecretEventSource;
+  }
+}