diff --git a/docs/content/en/docs/documentation/reconciler.md b/docs/content/en/docs/documentation/reconciler.md
index 7af6527422..a60ddd2b12 100644
--- a/docs/content/en/docs/documentation/reconciler.md
+++ b/docs/content/en/docs/documentation/reconciler.md
@@ -258,4 +258,63 @@ In this mode:
 - you cannot use managed dependent resources since those manage the finalizers and other logic related to the normal
   execution mode.
 
-See also [sample](https://github.com/operator-framework/java-operator-sdk/blob/main/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/triggerallevent/finalizerhandling) for selectively adding finalizers for resources;
\ No newline at end of file
+See also [sample](https://github.com/operator-framework/java-operator-sdk/blob/main/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/triggerallevent/finalizerhandling) for selectively adding finalizers for resources;
+
+### Expectations
+
+Expectations are a pattern to ensure that, during reconciliation, your secondary resources are in a certain state.
+For a more detailed explanation see [this blogpost](https://ahmet.im/blog/controller-pitfalls/#expectations-pattern).
+You can find framework support for this pattern in [`io.javaoperatorsdk.operator.processing.expectation`](https://github.com/operator-framework/java-operator-sdk/blob/main/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/expectation/) 
+package. See also related [integration test](https://github.com/operator-framework/java-operator-sdk/blob/main/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/ExpectationReconciler.java).
+Note that this feature is marked as `@Experimental`, since based on feedback the API might be improved / changed, but we intend 
+to support it, later also might be integrated to Dependent Resources and/or Workflows.
+
+The idea is the nutshell, is that you can track your expectations in the expectation manager in the reconciler
+which has an API that covers the common use cases. 
+
+The following sample is the simplified version of the integration test that implements the logic that creates a 
+deployment and sets status message if there are the target three replicas ready:
+
+```java
+public class ExpectationReconciler implements Reconciler<ExpectationCustomResource> {
+
+    // some code is omitted
+    
+    private final ExpectationManager<ExpectationCustomResource> expectationManager =
+            new ExpectationManager<>();
+
+    @Override
+    public UpdateControl<ExpectationCustomResource> reconcile(
+            ExpectationCustomResource primary, Context<ExpectationCustomResource> context) {
+
+        // exiting asap if there is an expectation that is not timed out neither fulfilled yet
+        if (expectationManager.ongoingExpectationPresent(primary, context)) {
+            return UpdateControl.noUpdate();
+        }
+
+        var deployment = context.getSecondaryResource(Deployment.class);
+        if (deployment.isEmpty()) {
+            createDeployment(primary, context);
+            expectationManager.setExpectation(
+                    primary, Duration.ofSeconds(timeout), deploymentReadyExpectation(context));
+            return UpdateControl.noUpdate();
+        } else {
+            // checks if the expectation if it is fulfilled, and also removes it.
+            // in your logic you might add a next expectation based on your workflow.
+            // Expectations have a name, so you can easily distinguish them if there is more of them.
+            var res = expectationManager.checkExpectation("deploymentReadyExpectation",primary, context);
+            if (res.isFulfilled()) {
+                return pathchStatusWithMessage(primary, DEPLOYMENT_READY);
+            } else if (res.isTimedOut()) {
+                // you might add some other timeout handling here
+                return pathchStatusWithMessage(primary, DEPLOYMENT_TIMEOUT);
+            }
+        }
+        return UpdateControl.noUpdate();
+        
+    }
+}
+```
+
+
+
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/Experimental.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/Experimental.java
index bd9fe596e6..963c56b47b 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/Experimental.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/Experimental.java
@@ -29,6 +29,11 @@
 @Retention(RetentionPolicy.SOURCE)
 @Target({ElementType.METHOD, ElementType.TYPE, ElementType.FIELD, ElementType.PACKAGE})
 public @interface Experimental {
+  /**
+   * Message for experimental features that we intend to keep and maintain, but the API might change
+   * usually, based on user feedback.
+   */
+  String API_MIGHT_CHANGE = "API might change, usually based on feedback";
 
   /**
    * Describes why the annotated element is experimental.
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/expectation/Expectation.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/expectation/Expectation.java
new file mode 100644
index 0000000000..f3d53c3188
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/expectation/Expectation.java
@@ -0,0 +1,51 @@
+/*
+ * Copyright Java Operator SDK Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.javaoperatorsdk.operator.processing.expectation;
+
+import java.util.function.BiPredicate;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.Experimental;
+
+import static io.javaoperatorsdk.operator.api.reconciler.Experimental.API_MIGHT_CHANGE;
+
+@Experimental(API_MIGHT_CHANGE)
+public interface Expectation<P extends HasMetadata> {
+
+  String UNNAMED = "unnamed";
+
+  boolean isFulfilled(P primary, Context<P> context);
+
+  default String name() {
+    return UNNAMED;
+  }
+
+  static <P extends HasMetadata> Expectation<P> createExpectation(
+      String name, BiPredicate<P, Context<P>> predicate) {
+    return new Expectation<>() {
+      @Override
+      public String name() {
+        return name;
+      }
+
+      @Override
+      public boolean isFulfilled(P primary, Context<P> context) {
+        return predicate.test(primary, context);
+      }
+    };
+  }
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/expectation/ExpectationManager.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/expectation/ExpectationManager.java
new file mode 100644
index 0000000000..1c2f9d8ed3
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/expectation/ExpectationManager.java
@@ -0,0 +1,143 @@
+/*
+ * Copyright Java Operator SDK Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.javaoperatorsdk.operator.processing.expectation;
+
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.Experimental;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+
+import static io.javaoperatorsdk.operator.api.reconciler.Experimental.API_MIGHT_CHANGE;
+
+@Experimental(API_MIGHT_CHANGE)
+public class ExpectationManager<P extends HasMetadata> {
+
+  protected final ConcurrentHashMap<ResourceID, RegisteredExpectation<P>> registeredExpectations =
+      new ConcurrentHashMap<>();
+
+  /**
+   * Checks if the expectation holds, if not sets the expectation with the given timeout.
+   *
+   * @return false, if the expectation is already fulfilled, therefore, not registered. Returns true
+   *     if expectation is not met and set with a timeout.
+   */
+  public boolean checkAndSetExpectation(
+      P primary, Context<P> context, Duration timeout, Expectation<P> expectation) {
+    var fulfilled = expectation.isFulfilled(primary, context);
+    if (fulfilled) {
+      return false;
+    } else {
+      setExpectation(primary, timeout, expectation);
+      return true;
+    }
+  }
+
+  /**
+   * Sets a target expectation with given timeout.
+   *
+   * @param primary resource
+   * @param timeout of expectation
+   * @param expectation to check
+   */
+  // we might consider in the future to throw an exception if an expectation is already set
+  public void setExpectation(P primary, Duration timeout, Expectation<P> expectation) {
+    registeredExpectations.put(
+        ResourceID.fromResource(primary),
+        new RegisteredExpectation<>(LocalDateTime.now(), timeout, expectation));
+  }
+
+  /**
+   * Checks on expectation with provided name. Return the expectation result. If the result of
+   * expectation is fulfilled, the expectation is automatically removed;
+   */
+  public ExpectationResult<P> checkExpectation(
+      String expectationName, P primary, Context<P> context) {
+    var resourceID = ResourceID.fromResource(primary);
+    var exp = registeredExpectations.get(ResourceID.fromResource(primary));
+    if (exp != null && expectationName.equals(exp.expectation().name())) {
+      return checkExpectation(exp, resourceID, primary, context);
+    } else {
+      return checkExpectation(null, resourceID, primary, context);
+    }
+  }
+
+  /**
+   * Checks if actual expectation is fulfilled. Return the expectation result. If the result of
+   * expectation is fulfilled, the expectation is automatically removed;
+   */
+  public ExpectationResult<P> checkExpectation(P primary, Context<P> context) {
+    var resourceID = ResourceID.fromResource(primary);
+    var exp = registeredExpectations.get(ResourceID.fromResource(primary));
+    return checkExpectation(exp, resourceID, primary, context);
+  }
+
+  private ExpectationResult<P> checkExpectation(
+      RegisteredExpectation<P> exp, ResourceID resourceID, P primary, Context<P> context) {
+    if (exp == null) {
+      return new ExpectationResult<>(null, null);
+    }
+    if (exp.expectation().isFulfilled(primary, context)) {
+      registeredExpectations.remove(resourceID);
+      return new ExpectationResult<>(exp.expectation(), ExpectationStatus.FULFILLED);
+    } else if (exp.isTimedOut()) {
+      // we don't remove the expectation so user knows about it's state
+      return new ExpectationResult<>(exp.expectation(), ExpectationStatus.TIMED_OUT);
+    } else {
+      return new ExpectationResult<>(exp.expectation(), ExpectationStatus.NOT_YET_FULFILLED);
+    }
+  }
+
+  /*
+   * Returns true if there is an expectation for the primary resource, but it is not yet fulfilled
+   * neither timed out.
+   * The intention behind is that you can exit reconciliation early with a simple check
+   * if true.
+   * */
+  public boolean ongoingExpectationPresent(P primary, Context<P> context) {
+    var exp = registeredExpectations.get(ResourceID.fromResource(primary));
+    if (exp == null) {
+      return false;
+    }
+    return !exp.isTimedOut() && !exp.expectation().isFulfilled(primary, context);
+  }
+
+  public boolean isExpectationPresent(P primary) {
+    return registeredExpectations.containsKey(ResourceID.fromResource(primary));
+  }
+
+  public boolean isExpectationPresent(String name, P primary) {
+    var exp = registeredExpectations.get(ResourceID.fromResource(primary));
+    return exp != null && name.equals(exp.expectation().name());
+  }
+
+  public Optional<Expectation<P>> getExpectation(P primary) {
+    var regExp = registeredExpectations.get(ResourceID.fromResource(primary));
+    return Optional.ofNullable(regExp).map(RegisteredExpectation::expectation);
+  }
+
+  public Optional<String> getExpectationName(P primary) {
+    return getExpectation(primary).map(Expectation::name);
+  }
+
+  public void removeExpectation(P primary) {
+    registeredExpectations.remove(ResourceID.fromResource(primary));
+  }
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/expectation/ExpectationResult.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/expectation/ExpectationResult.java
new file mode 100644
index 0000000000..5b4081f7b3
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/expectation/ExpectationResult.java
@@ -0,0 +1,42 @@
+/*
+ * Copyright Java Operator SDK Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.javaoperatorsdk.operator.processing.expectation;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+
+public record ExpectationResult<P extends HasMetadata>(
+    Expectation<P> expectation, ExpectationStatus status) {
+
+  public boolean isFulfilled() {
+    return status == ExpectationStatus.FULFILLED;
+  }
+
+  public boolean isTimedOut() {
+    return status == ExpectationStatus.TIMED_OUT;
+  }
+
+  public boolean isExpectationPresent() {
+    return expectation != null;
+  }
+
+  public boolean isNotPresentOrFulfilled() {
+    return !isExpectationPresent() || isFulfilled();
+  }
+
+  public String name() {
+    return expectation.name();
+  }
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/expectation/ExpectationStatus.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/expectation/ExpectationStatus.java
new file mode 100644
index 0000000000..7b6df29fad
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/expectation/ExpectationStatus.java
@@ -0,0 +1,22 @@
+/*
+ * Copyright Java Operator SDK Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.javaoperatorsdk.operator.processing.expectation;
+
+public enum ExpectationStatus {
+  FULFILLED,
+  NOT_YET_FULFILLED,
+  TIMED_OUT
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/expectation/PeriodicCleanerExpectationManager.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/expectation/PeriodicCleanerExpectationManager.java
new file mode 100644
index 0000000000..d2145e2afe
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/expectation/PeriodicCleanerExpectationManager.java
@@ -0,0 +1,69 @@
+/*
+ * Copyright Java Operator SDK Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.javaoperatorsdk.operator.processing.expectation;
+
+import java.time.Duration;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.Experimental;
+import io.javaoperatorsdk.operator.api.reconciler.IndexedResourceCache;
+
+import static io.javaoperatorsdk.operator.api.reconciler.Experimental.API_MIGHT_CHANGE;
+
+/**
+ * Expectation manager implementation that works without enabling {@link
+ * ControllerConfiguration#triggerReconcilerOnAllEvent()}. Periodically checks and cleanups'
+ * expectations for primary resources which are no longer present in the cache.
+ */
+@Experimental(API_MIGHT_CHANGE)
+public class PeriodicCleanerExpectationManager<P extends HasMetadata>
+    extends ExpectationManager<P> {
+
+  public static final Duration DEFAULT_CHECK_PERIOD = Duration.ofMinutes(1);
+
+  private final ScheduledExecutorService scheduler =
+      Executors.newScheduledThreadPool(
+          1,
+          r -> {
+            Thread thread = Executors.defaultThreadFactory().newThread(r);
+            thread.setDaemon(true);
+            return thread;
+          });
+
+  private final IndexedResourceCache<P> primaryCache;
+
+  public PeriodicCleanerExpectationManager(IndexedResourceCache<P> primaryCache) {
+    this(DEFAULT_CHECK_PERIOD, primaryCache);
+  }
+
+  public PeriodicCleanerExpectationManager(Duration period, IndexedResourceCache<P> primaryCache) {
+    this.primaryCache = primaryCache;
+    scheduler.scheduleWithFixedDelay(
+        this::clean, period.toMillis(), period.toMillis(), TimeUnit.MICROSECONDS);
+  }
+
+  public void clean() {
+    registeredExpectations.entrySet().removeIf(e -> primaryCache.get(e.getKey()).isEmpty());
+  }
+
+  void stop() {
+    scheduler.shutdownNow();
+  }
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/expectation/RegisteredExpectation.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/expectation/RegisteredExpectation.java
new file mode 100644
index 0000000000..b22a810d6c
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/expectation/RegisteredExpectation.java
@@ -0,0 +1,29 @@
+/*
+ * Copyright Java Operator SDK Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.javaoperatorsdk.operator.processing.expectation;
+
+import java.time.Duration;
+import java.time.LocalDateTime;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+
+record RegisteredExpectation<P extends HasMetadata>(
+    LocalDateTime registeredAt, Duration timeout, Expectation<P> expectation) {
+
+  public boolean isTimedOut() {
+    return LocalDateTime.now().isAfter(registeredAt.plus(timeout));
+  }
+}
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/expectation/ExpectationManagerTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/expectation/ExpectationManagerTest.java
new file mode 100644
index 0000000000..1a69e4057b
--- /dev/null
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/expectation/ExpectationManagerTest.java
@@ -0,0 +1,208 @@
+/*
+ * Copyright Java Operator SDK Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.javaoperatorsdk.operator.processing.expectation;
+
+import java.time.Duration;
+import java.util.Optional;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class ExpectationManagerTest {
+
+  private ExpectationManager<ConfigMap> expectationManager;
+  private ConfigMap configMap;
+  private Context<ConfigMap> context;
+
+  @BeforeEach
+  void setUp() {
+    expectationManager = new ExpectationManager<>();
+    configMap = new ConfigMap();
+    configMap.setMetadata(
+        new ObjectMetaBuilder().withName("test-configmap").withNamespace("test-namespace").build());
+    context = mock(Context.class);
+  }
+
+  @Test
+  void setExpectationShouldStoreExpectation() {
+    Expectation<ConfigMap> expectation = mock(Expectation.class);
+    Duration timeout = Duration.ofMinutes(5);
+
+    expectationManager.setExpectation(configMap, timeout, expectation);
+
+    assertThat(expectationManager.isExpectationPresent(configMap)).isTrue();
+    assertThat(expectationManager.getExpectation(configMap)).contains(expectation);
+  }
+
+  @Test
+  void checkExpectationShouldReturnEmptyWhenNoExpectation() {
+    ExpectationResult<ConfigMap> result = expectationManager.checkExpectation(configMap, context);
+
+    assertThat(result.isExpectationPresent()).isFalse();
+  }
+
+  @Test
+  void checkExpectationShouldReturnFulfilledWhenExpectationMet() {
+    Expectation<ConfigMap> expectation = mock(Expectation.class);
+    when(expectation.isFulfilled(configMap, context)).thenReturn(true);
+
+    expectationManager.setExpectation(configMap, Duration.ofMinutes(5), expectation);
+    ExpectationResult<ConfigMap> result = expectationManager.checkExpectation(configMap, context);
+
+    assertThat(result.isExpectationPresent()).isTrue();
+    assertThat(result.isFulfilled()).isTrue();
+    assertThat(result.expectation()).isEqualTo(expectation);
+    assertThat(expectationManager.isExpectationPresent(configMap)).isFalse();
+  }
+
+  @Test
+  void checkExpectationShouldReturnNotFulfilledWhenExpectationNotMet() {
+    Expectation<ConfigMap> expectation = mock(Expectation.class);
+    when(expectation.isFulfilled(configMap, context)).thenReturn(false);
+
+    expectationManager.setExpectation(configMap, Duration.ofMinutes(5), expectation);
+    ExpectationResult<ConfigMap> result = expectationManager.checkExpectation(configMap, context);
+
+    assertThat(result.isExpectationPresent()).isTrue();
+    assertThat(result.isFulfilled()).isFalse();
+    assertThat(result.expectation()).isEqualTo(expectation);
+    assertThat(expectationManager.isExpectationPresent(configMap)).isTrue();
+  }
+
+  @Test
+  void checkExpectationShouldReturnTimedOutWhenExpectationExpired() throws InterruptedException {
+    Expectation<ConfigMap> expectation = mock(Expectation.class);
+    when(expectation.isFulfilled(configMap, context)).thenReturn(false);
+
+    expectationManager.setExpectation(configMap, Duration.ofMillis(1), expectation);
+    Thread.sleep(10);
+    ExpectationResult<ConfigMap> result = expectationManager.checkExpectation(configMap, context);
+
+    assertThat(result.isExpectationPresent()).isTrue();
+    assertThat(result.isTimedOut()).isTrue();
+    assertThat(expectationManager.isExpectationPresent(configMap)).isTrue();
+  }
+
+  @Test
+  void getExpectationNameShouldReturnExpectationName() {
+    String expectedName = "test-expectation";
+    Expectation<ConfigMap> expectation = mock(Expectation.class);
+    when(expectation.name()).thenReturn(expectedName);
+
+    expectationManager.setExpectation(configMap, Duration.ofMinutes(5), expectation);
+    Optional<String> name = expectationManager.getExpectationName(configMap);
+
+    assertThat(name).contains(expectedName);
+  }
+
+  @Test
+  void getExpectationNameShouldReturnEmptyWhenNoExpectation() {
+    Optional<String> name = expectationManager.getExpectationName(configMap);
+
+    assertThat(name).isEmpty();
+  }
+
+  @Test
+  void removeExpectationShouldRemoveExpectation() {
+    Expectation<ConfigMap> expectation = mock(Expectation.class);
+
+    expectationManager.setExpectation(configMap, Duration.ofMinutes(5), expectation);
+    assertThat(expectationManager.isExpectationPresent(configMap)).isTrue();
+
+    expectationManager.removeExpectation(configMap);
+    assertThat(expectationManager.isExpectationPresent(configMap)).isFalse();
+  }
+
+  @Test
+  void checkingSpecificExpectation() {
+    String expectedName = "test-expectation";
+    Expectation<ConfigMap> expectation = mock(Expectation.class);
+    when(expectation.name()).thenReturn(expectedName);
+    when(expectation.isFulfilled(any(), any())).thenReturn(true);
+
+    expectationManager.setExpectation(configMap, Duration.ofMinutes(1), expectation);
+
+    var res = expectationManager.checkExpectation("other-expectation", configMap, context);
+    assertThat(res.isExpectationPresent()).isFalse();
+    res = expectationManager.checkExpectation(expectedName, configMap, context);
+    assertThat(res.isExpectationPresent()).isTrue();
+  }
+
+  @Test
+  void shouldHandleMultipleExpectationsForDifferentResources() {
+    ConfigMap configMap2 = new ConfigMap();
+    configMap2.setMetadata(
+        new ObjectMetaBuilder()
+            .withName("test-configmap-2")
+            .withNamespace("test-namespace")
+            .build());
+
+    Expectation<ConfigMap> expectation1 = mock(Expectation.class);
+    Expectation<ConfigMap> expectation2 = mock(Expectation.class);
+
+    expectationManager.setExpectation(configMap, Duration.ofMinutes(5), expectation1);
+    expectationManager.setExpectation(configMap2, Duration.ofMinutes(5), expectation2);
+
+    assertThat(expectationManager.isExpectationPresent(configMap)).isTrue();
+    assertThat(expectationManager.isExpectationPresent(configMap2)).isTrue();
+    assertThat(expectationManager.getExpectation(configMap)).contains(expectation1);
+    assertThat(expectationManager.getExpectation(configMap2)).contains(expectation2);
+  }
+
+  @Test
+  void setExpectationShouldReplaceExistingExpectation() {
+    Expectation<ConfigMap> expectation1 = mock(Expectation.class);
+    Expectation<ConfigMap> expectation2 = mock(Expectation.class);
+
+    expectationManager.setExpectation(configMap, Duration.ofMinutes(5), expectation1);
+    expectationManager.setExpectation(configMap, Duration.ofMinutes(5), expectation2);
+
+    assertThat(expectationManager.getExpectation(configMap)).contains(expectation2);
+  }
+
+  @Test
+  void checkAndSetExpectationAlreadyMet() {
+    Expectation<ConfigMap> expectation = mock(Expectation.class);
+    when(expectation.isFulfilled(any(), any())).thenReturn(true);
+
+    var res =
+        expectationManager.checkAndSetExpectation(
+            configMap, mock(Context.class), Duration.ofMinutes(5), expectation);
+    assertThat(res).isFalse();
+    assertThat(expectationManager.getExpectation(configMap)).isEmpty();
+  }
+
+  @Test
+  void checkAndSetExpectationNotMet() {
+    Expectation<ConfigMap> expectation = mock(Expectation.class);
+    when(expectation.isFulfilled(any(), any())).thenReturn(false);
+
+    var res =
+        expectationManager.checkAndSetExpectation(
+            configMap, mock(Context.class), Duration.ofMinutes(5), expectation);
+    assertThat(res).isTrue();
+    assertThat(expectationManager.getExpectation(configMap)).isPresent();
+  }
+}
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/expectation/PeriodicCleanerExpectationManagerTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/expectation/PeriodicCleanerExpectationManagerTest.java
new file mode 100644
index 0000000000..a317d1aaa7
--- /dev/null
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/expectation/PeriodicCleanerExpectationManagerTest.java
@@ -0,0 +1,96 @@
+/*
+ * Copyright Java Operator SDK Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.javaoperatorsdk.operator.processing.expectation;
+
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.javaoperatorsdk.operator.api.reconciler.IndexedResourceCache;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+import static org.mockito.Mockito.when;
+
+class PeriodicCleanerExpectationManagerTest {
+
+  @Mock private IndexedResourceCache<ConfigMap> primaryCache;
+
+  private PeriodicCleanerExpectationManager<ConfigMap> expectationManager;
+  private ConfigMap configMap;
+  private AutoCloseable closeable;
+
+  @BeforeEach
+  void setUp() {
+    closeable = MockitoAnnotations.openMocks(this);
+    configMap = new ConfigMap();
+    configMap.setMetadata(
+        new ObjectMetaBuilder().withName("test-configmap").withNamespace("test-namespace").build());
+  }
+
+  @AfterEach
+  void tearDown() throws Exception {
+    if (expectationManager != null) {
+      expectationManager.stop();
+    }
+    closeable.close();
+  }
+
+  @Test
+  void shouldCleanExpectationsWhenResourceNotInCache() {
+    Duration period = Duration.ofMillis(50);
+    expectationManager = new PeriodicCleanerExpectationManager<>(period, primaryCache);
+
+    ResourceID resourceId = ResourceID.fromResource(configMap);
+    when(primaryCache.get(resourceId)).thenReturn(java.util.Optional.empty());
+
+    Expectation<ConfigMap> expectation = (primary, context) -> false;
+    expectationManager.setExpectation(configMap, Duration.ofMinutes(10), expectation);
+
+    assertThat(expectationManager.isExpectationPresent(configMap)).isTrue();
+
+    await()
+        .atMost(200, TimeUnit.MILLISECONDS)
+        .untilAsserted(
+            () -> assertThat(expectationManager.isExpectationPresent(configMap)).isFalse());
+  }
+
+  @Test
+  void shouldNotCleanExpectationsWhenResourceInCache() throws InterruptedException {
+    Duration period = Duration.ofMillis(50);
+    expectationManager = new PeriodicCleanerExpectationManager<>(period, primaryCache);
+
+    ResourceID resourceId = ResourceID.fromResource(configMap);
+    when(primaryCache.get(resourceId)).thenReturn(java.util.Optional.of(configMap));
+
+    Expectation<ConfigMap> expectation = (primary, context) -> false;
+    expectationManager.setExpectation(configMap, Duration.ofMinutes(10), expectation);
+
+    assertThat(expectationManager.isExpectationPresent(configMap)).isTrue();
+
+    Thread.sleep(150);
+
+    assertThat(expectationManager.isExpectationPresent(configMap)).isTrue();
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/ExpectationCustomResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/ExpectationCustomResource.java
new file mode 100644
index 0000000000..4568550b9e
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/ExpectationCustomResource.java
@@ -0,0 +1,28 @@
+/*
+ * Copyright Java Operator SDK Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.javaoperatorsdk.operator.baseapi.expectation;
+
+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("ecr")
+public class ExpectationCustomResource extends CustomResource<Void, ExpectationCustomResourceStatus>
+    implements Namespaced {}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/ExpectationCustomResourceStatus.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/ExpectationCustomResourceStatus.java
new file mode 100644
index 0000000000..f6ae538d4f
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/ExpectationCustomResourceStatus.java
@@ -0,0 +1,29 @@
+/*
+ * Copyright Java Operator SDK Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.javaoperatorsdk.operator.baseapi.expectation;
+
+public class ExpectationCustomResourceStatus {
+
+  private String message;
+
+  public String getMessage() {
+    return message;
+  }
+
+  public void setMessage(String message) {
+    this.message = message;
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/ExpectationIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/ExpectationIT.java
new file mode 100644
index 0000000000..234f3b3949
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/ExpectationIT.java
@@ -0,0 +1,71 @@
+/*
+ * Copyright Java Operator SDK Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.javaoperatorsdk.operator.baseapi.expectation;
+
+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;
+
+import static io.javaoperatorsdk.operator.baseapi.expectation.ExpectationReconciler.DEPLOYMENT_READY;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+class ExpectationIT {
+
+  public static final String TEST_1 = "test1";
+
+  @RegisterExtension
+  LocallyRunOperatorExtension extension =
+      LocallyRunOperatorExtension.builder().withReconciler(new ExpectationReconciler()).build();
+
+  @Test
+  void testExpectation() {
+    extension.getReconcilerOfType(ExpectationReconciler.class).setTimeout(30000L);
+    var res = testResource();
+    extension.create(res);
+
+    await()
+        .untilAsserted(
+            () -> {
+              var actual = extension.get(ExpectationCustomResource.class, TEST_1);
+              assertThat(actual.getStatus()).isNotNull();
+              assertThat(actual.getStatus().getMessage()).isEqualTo(DEPLOYMENT_READY);
+            });
+  }
+
+  @Test
+  void expectationTimeouts() {
+    extension.getReconcilerOfType(ExpectationReconciler.class).setTimeout(300L);
+    var res = testResource();
+    extension.create(res);
+
+    await()
+        .untilAsserted(
+            () -> {
+              var actual = extension.get(ExpectationCustomResource.class, TEST_1);
+              assertThat(actual.getStatus()).isNotNull();
+              assertThat(actual.getStatus().getMessage()).isEqualTo(DEPLOYMENT_READY);
+            });
+  }
+
+  private ExpectationCustomResource testResource() {
+    var res = new ExpectationCustomResource();
+    res.setMetadata(new ObjectMetaBuilder().withName(TEST_1).build());
+    return res;
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/ExpectationReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/ExpectationReconciler.java
new file mode 100644
index 0000000000..3a57d3effd
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/expectation/ExpectationReconciler.java
@@ -0,0 +1,167 @@
+/*
+ * Copyright Java Operator SDK Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.javaoperatorsdk.operator.baseapi.expectation;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+
+import io.fabric8.kubernetes.api.model.ContainerBuilder;
+import io.fabric8.kubernetes.api.model.ContainerPortBuilder;
+import io.fabric8.kubernetes.api.model.LabelSelectorBuilder;
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.fabric8.kubernetes.api.model.PodSpecBuilder;
+import io.fabric8.kubernetes.api.model.PodTemplateSpecBuilder;
+import io.fabric8.kubernetes.api.model.apps.Deployment;
+import io.fabric8.kubernetes.api.model.apps.DeploymentBuilder;
+import io.fabric8.kubernetes.api.model.apps.DeploymentSpecBuilder;
+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.processing.expectation.Expectation;
+import io.javaoperatorsdk.operator.processing.expectation.ExpectationManager;
+
+@ControllerConfiguration(triggerReconcilerOnAllEvent = true)
+public class ExpectationReconciler implements Reconciler<ExpectationCustomResource> {
+
+  public static final String DEPLOYMENT_READY = "Deployment ready";
+  public static final String DEPLOYMENT_TIMEOUT = "Deployment timeout";
+  public static final String DEPLOYMENT_READY_EXPECTATION_NAME = "deploymentReadyExpectation";
+  private final ExpectationManager<ExpectationCustomResource> expectationManager =
+      new ExpectationManager<>();
+
+  private volatile Long timeout = 30000l;
+
+  @Override
+  public UpdateControl<ExpectationCustomResource> reconcile(
+      ExpectationCustomResource primary, Context<ExpectationCustomResource> context) {
+
+    // cleans up expectation manager for the resource on delete event
+    // in case of cleaner interface used, this can done also there.
+    if (context.isPrimaryResourceDeleted()) {
+      expectationManager.removeExpectation(primary);
+    }
+
+    // exiting asap if there is an expectation that is not timed out neither fulfilled
+    if (expectationManager.ongoingExpectationPresent(primary, context)) {
+      return UpdateControl.noUpdate();
+    }
+
+    var deployment = context.getSecondaryResource(Deployment.class);
+    if (deployment.isEmpty()) {
+      createDeployment(primary, context);
+      var set =
+          expectationManager.checkAndSetExpectation(
+              primary, context, Duration.ofSeconds(timeout), deploymentReadyExpectation(context));
+      if (set) {
+        return UpdateControl.noUpdate();
+      }
+    } else {
+      // checks the expectation if it is fulfilled also removes it,
+      // in your logic you might add a next expectation based on your workflow
+      // Expectations have a name, so you can easily distinguish them if there is more of them.
+      var res =
+          expectationManager.checkExpectation(DEPLOYMENT_READY_EXPECTATION_NAME, primary, context);
+      // Note that this happens only once, since if the expectation is fulfilled, it is also removed
+      // from the manager.
+      if (res.isFulfilled()) {
+        return pathchStatusWithMessage(primary, DEPLOYMENT_READY);
+      } else if (res.isTimedOut()) {
+        // you might add some other timeout handling here
+        return pathchStatusWithMessage(primary, DEPLOYMENT_TIMEOUT);
+      }
+    }
+    return UpdateControl.noUpdate();
+  }
+
+  private static UpdateControl<ExpectationCustomResource> pathchStatusWithMessage(
+      ExpectationCustomResource primary, String message) {
+    primary.setStatus(new ExpectationCustomResourceStatus());
+    primary.getStatus().setMessage(message);
+    return UpdateControl.patchStatus(primary);
+  }
+
+  private static Expectation<ExpectationCustomResource> deploymentReadyExpectation(
+      Context<ExpectationCustomResource> context) {
+    return Expectation.createExpectation(
+        DEPLOYMENT_READY_EXPECTATION_NAME,
+        (p, c) ->
+            context
+                .getSecondaryResource(Deployment.class)
+                .map(
+                    ad ->
+                        ad.getStatus() != null
+                            && ad.getStatus().getReadyReplicas() != null
+                            && ad.getStatus().getReadyReplicas() == 3)
+                .orElse(false));
+  }
+
+  private Deployment createDeployment(
+      ExpectationCustomResource primary, Context<ExpectationCustomResource> context) {
+    var d =
+        new DeploymentBuilder()
+            .withMetadata(
+                new ObjectMetaBuilder()
+                    .withName(primary.getMetadata().getName())
+                    .withNamespace(primary.getMetadata().getNamespace())
+                    .build())
+            .withSpec(
+                new DeploymentSpecBuilder()
+                    .withReplicas(3)
+                    .withSelector(
+                        new LabelSelectorBuilder().withMatchLabels(Map.of("app", "nginx")).build())
+                    .withTemplate(
+                        new PodTemplateSpecBuilder()
+                            .withMetadata(
+                                new ObjectMetaBuilder().withLabels(Map.of("app", "nginx")).build())
+                            .withSpec(
+                                new PodSpecBuilder()
+                                    .withContainers(
+                                        new ContainerBuilder()
+                                            .withName("nginx")
+                                            .withImage("nginx:1.29.2")
+                                            .withPorts(
+                                                new ContainerPortBuilder()
+                                                    .withContainerPort(80)
+                                                    .build())
+                                            .build())
+                                    .build())
+                            .build())
+                    .build())
+            .build();
+    d.addOwnerReference(primary);
+    return context.getClient().resource(d).serverSideApply();
+  }
+
+  @Override
+  public List<EventSource<?, ExpectationCustomResource>> prepareEventSources(
+      EventSourceContext<ExpectationCustomResource> context) {
+    return List.of(
+        new InformerEventSource<>(
+            InformerEventSourceConfiguration.from(Deployment.class, ExpectationCustomResource.class)
+                .build(),
+            context));
+  }
+
+  public void setTimeout(Long timeout) {
+    this.timeout = timeout;
+  }
+}