diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractDependentResource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractDependentResource.java
index 2c8e5af017..dbfc34fe42 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractDependentResource.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractDependentResource.java
@@ -110,10 +110,9 @@ private void logForOperation(String operation, P primary, R desired) {
   }
 
   protected R handleCreate(R desired, P primary, Context<P> context) {
-    ResourceID resourceID = ResourceID.fromResource(primary);
     R created = creator.create(desired, primary, context);
     throwIfNull(created, primary, "Created resource");
-    onCreated(resourceID, created);
+    onCreated(primary, created, context);
     return created;
   }
 
@@ -121,27 +120,28 @@ protected R handleCreate(R desired, P primary, Context<P> context) {
    * Allows subclasses to perform additional processing (e.g. caching) on the created resource if
    * needed.
    *
-   * @param primaryResourceId the {@link ResourceID} of the primary resource associated with the
-   *        newly created resource
+   * @param primary the {@link ResourceID} of the primary resource associated with the newly created
+   *        resource
    * @param created the newly created resource
+   * @param context
    */
-  protected abstract void onCreated(ResourceID primaryResourceId, R created);
+  protected abstract void onCreated(P primary, R created, Context<P> context);
 
   /**
    * Allows subclasses to perform additional processing on the updated resource if needed.
    *
-   * @param primaryResourceId the {@link ResourceID} of the primary resource associated with the
-   *        newly updated resource
+   * @param primary the {@link ResourceID} of the primary resource associated with the newly updated
+   *        resource
    * @param updated the updated resource
    * @param actual the resource as it was before the update
+   * @param context
    */
-  protected abstract void onUpdated(ResourceID primaryResourceId, R updated, R actual);
+  protected abstract void onUpdated(P primary, R updated, R actual, Context<P> context);
 
   protected R handleUpdate(R actual, R desired, P primary, Context<P> context) {
-    ResourceID resourceID = ResourceID.fromResource(primary);
     R updated = updater.update(actual, desired, primary, context);
     throwIfNull(updated, primary, "Updated resource");
-    onUpdated(resourceID, updated, actual);
+    onUpdated(primary, updated, actual, context);
     return updated;
   }
 
@@ -154,8 +154,9 @@ public void delete(P primary, Context<P> context) {
     dependentResourceReconciler.delete(primary, context);
   }
 
-  protected void handleDelete(P primary, Context<P> context) {
-    throw new IllegalStateException("delete method be implemented if Deleter trait is supported");
+  protected void handleDelete(P primary, R secondary, Context<P> context) {
+    throw new IllegalStateException(
+        "handleDelete method must be implemented if Deleter trait is supported");
   }
 
   public void setResourceDiscriminator(
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractEventSourceHolderDependentResource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractEventSourceHolderDependentResource.java
index f514a00950..ad1eeb4d43 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractEventSourceHolderDependentResource.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractEventSourceHolderDependentResource.java
@@ -3,6 +3,7 @@
 import java.util.Optional;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
 import io.javaoperatorsdk.operator.api.reconciler.Ignore;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.EventSourceNotFoundException;
@@ -94,15 +95,17 @@ public Optional<ResourceEventSource<R, P>> eventSource() {
     return Optional.ofNullable(eventSource);
   }
 
-  protected void onCreated(ResourceID primaryResourceId, R created) {
+  protected void onCreated(P primary, R created, Context<P> context) {
     if (isCacheFillerEventSource) {
-      recentOperationCacheFiller().handleRecentResourceCreate(primaryResourceId, created);
+      recentOperationCacheFiller().handleRecentResourceCreate(ResourceID.fromResource(primary),
+          created);
     }
   }
 
-  protected void onUpdated(ResourceID primaryResourceId, R updated, R actual) {
+  protected void onUpdated(P primary, R updated, R actual, Context<P> context) {
     if (isCacheFillerEventSource) {
-      recentOperationCacheFiller().handleRecentResourceUpdate(primaryResourceId, updated, actual);
+      recentOperationCacheFiller().handleRecentResourceUpdate(ResourceID.fromResource(primary),
+          updated, actual);
     }
   }
 
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractExternalDependentResource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractExternalDependentResource.java
new file mode 100644
index 0000000000..e3dd9c630a
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractExternalDependentResource.java
@@ -0,0 +1,120 @@
+package io.javaoperatorsdk.operator.processing.dependent;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.RecentOperationCacheFiller;
+import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+import io.javaoperatorsdk.operator.processing.event.source.ResourceEventSource;
+import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
+
+public abstract class AbstractExternalDependentResource<R, P extends HasMetadata, T extends ResourceEventSource<R, P>>
+    extends AbstractEventSourceHolderDependentResource<R, P, T> {
+
+  private final boolean isDependentResourceWithExplicitState =
+      this instanceof DependentResourceWithExplicitState;
+  private final boolean isBulkDependentResource = this instanceof BulkDependentResource;
+  @SuppressWarnings("rawtypes")
+  private DependentResourceWithExplicitState dependentResourceWithExplicitState;
+  private InformerEventSource<?, P> externalStateEventSource;
+  private KubernetesClient kubernetesClient;
+
+  @SuppressWarnings("unchecked")
+  protected AbstractExternalDependentResource(Class<R> resourceType) {
+    super(resourceType);
+    if (isDependentResourceWithExplicitState) {
+      dependentResourceWithExplicitState = (DependentResourceWithExplicitState<R, P, ?>) this;
+    }
+  }
+
+  @Override
+  @SuppressWarnings("unchecked")
+  public void resolveEventSource(EventSourceRetriever<P> eventSourceRetriever) {
+    super.resolveEventSource(eventSourceRetriever);
+    if (isDependentResourceWithExplicitState) {
+      externalStateEventSource = (InformerEventSource<?, P>) dependentResourceWithExplicitState
+          .eventSourceName()
+          .map(n -> eventSourceRetriever
+              .getResourceEventSourceFor(dependentResourceWithExplicitState.stateResourceClass(),
+                  (String) n))
+          .orElseGet(() -> eventSourceRetriever
+              .getResourceEventSourceFor(
+                  (Class<R>) dependentResourceWithExplicitState.stateResourceClass()));
+    }
+
+  }
+
+  @Override
+  protected void onCreated(P primary, R created, Context<P> context) {
+    super.onCreated(primary, created, context);
+    if (this instanceof DependentResourceWithExplicitState) {
+      handleExplicitStateCreation(primary, created, context);
+    }
+  }
+
+  @Override
+  public void delete(P primary, Context<P> context) {
+    if (isDependentResourceWithExplicitState && !isBulkDependentResource) {
+      var secondary = getSecondaryResource(primary, context);
+      super.delete(primary, context);
+      // deletes the state after the resource is deleted
+      handleExplicitStateDelete(primary, secondary.orElse(null), context);
+    } else {
+      super.delete(primary, context);
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private void handleExplicitStateDelete(P primary, R secondary, Context<P> context) {
+    var res = dependentResourceWithExplicitState.stateResource(primary, secondary);
+    dependentResourceWithExplicitState.getKubernetesClient().resource(res).delete();
+  }
+
+  @SuppressWarnings({"rawtypes", "unchecked"})
+  protected void handleExplicitStateCreation(P primary, R created, Context<P> context) {
+    var resource = dependentResourceWithExplicitState.stateResource(primary, created);
+    var stateResource =
+        dependentResourceWithExplicitState.getKubernetesClient().resource(resource).create();
+    if (externalStateEventSource != null) {
+      ((RecentOperationCacheFiller) externalStateEventSource)
+          .handleRecentResourceCreate(ResourceID.fromResource(primary), stateResource);
+    }
+  }
+
+
+  @SuppressWarnings("unchecked")
+  public void deleteTargetResource(P primary, R resource, String key,
+      Context<P> context) {
+    if (isDependentResourceWithExplicitState) {
+      getKubernetesClient()
+          .resource(dependentResourceWithExplicitState.stateResource(primary, resource))
+          .delete();
+    }
+    handleDeleteTargetResource(primary, resource, key, context);
+  }
+
+  public void handleDeleteTargetResource(P primary, R resource, String key,
+      Context<P> context) {
+    throw new IllegalStateException("Override this method in case you manage an bulk resource");
+  }
+
+  @SuppressWarnings("rawtypes")
+  protected InformerEventSource getExternalStateEventSource() {
+    return externalStateEventSource;
+  }
+
+  /**
+   * It's here just to manage the explicit state resource in case the dependent resource implements
+   * {@link RecentOperationCacheFiller}.
+   *
+   * @return kubernetes client.
+   */
+  public KubernetesClient getKubernetesClient() {
+    return kubernetesClient;
+  }
+
+  public void setKubernetesClient(KubernetesClient kubernetesClient) {
+    this.kubernetesClient = kubernetesClient;
+  }
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/BulkDependentResourceReconciler.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/BulkDependentResourceReconciler.java
index c6633f9db6..49b5af0ad4 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/BulkDependentResourceReconciler.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/BulkDependentResourceReconciler.java
@@ -11,7 +11,6 @@
 import io.javaoperatorsdk.operator.api.reconciler.dependent.Deleter;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.ReconcileResult;
 import io.javaoperatorsdk.operator.processing.dependent.Matcher.Result;
-import io.javaoperatorsdk.operator.processing.event.ResourceID;
 
 class BulkDependentResourceReconciler<R, P extends HasMetadata>
     implements DependentResourceReconciler<R, P> {
@@ -97,13 +96,13 @@ public Result<R> match(R resource, P primary, Context<P> context) {
     }
 
     @Override
-    protected void onCreated(ResourceID primaryResourceId, R created) {
-      asAbstractDependentResource().onCreated(primaryResourceId, created);
+    protected void onCreated(P primary, R created, Context<P> context) {
+      asAbstractDependentResource().onCreated(primary, created, context);
     }
 
     @Override
-    protected void onUpdated(ResourceID primaryResourceId, R updated, R actual) {
-      asAbstractDependentResource().onUpdated(primaryResourceId, updated, actual);
+    protected void onUpdated(P primary, R updated, R actual, Context<P> context) {
+      asAbstractDependentResource().onUpdated(primary, updated, actual, context);
     }
 
     @Override
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/DependentResourceWithExplicitState.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/DependentResourceWithExplicitState.java
new file mode 100644
index 0000000000..2a1b0a64c7
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/DependentResourceWithExplicitState.java
@@ -0,0 +1,37 @@
+package io.javaoperatorsdk.operator.processing.dependent;
+
+import java.util.Optional;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.Deleter;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.KubernetesClientAware;
+
+/**
+ * Handles external resources where in order to address the resource additional information or
+ * persistent state (usually the ID of the resource) is needed to access the current state. These
+ * are non Kubernetes resources which when created their ID is generated, so cannot be determined
+ * based only on primary resources. In order to manage such dependent resource use this interface
+ * for a resource that extends {@link AbstractExternalDependentResource}.
+ */
+public interface DependentResourceWithExplicitState<R, P extends HasMetadata, S extends HasMetadata>
+    extends Creator<R, P>, Deleter<P>, KubernetesClientAware {
+
+  /**
+   * Only needs to be implemented if multiple event sources are present for the target resource
+   * class.
+   *
+   * @return name of the event source to access the state resources.
+   */
+  default Optional<String> eventSourceName() {
+    return Optional.empty();
+  }
+
+  /**
+   * Class of the state resource.
+   */
+  Class<S> stateResourceClass();
+
+  /** State resource which contains the target state. Usually an ID to address the resource */
+  S stateResource(P primary, R resource);
+
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/SingleDependentResourceReconciler.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/SingleDependentResourceReconciler.java
index 20b37b3c7e..2862170c22 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/SingleDependentResourceReconciler.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/SingleDependentResourceReconciler.java
@@ -21,6 +21,7 @@ public ReconcileResult<R> reconcile(P primary, Context<P> context) {
 
   @Override
   public void delete(P primary, Context<P> context) {
-    instance.handleDelete(primary, context);
+    var secondary = instance.getSecondaryResource(primary, context);
+    instance.handleDelete(primary, secondary.orElse(null), context);
   }
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractPollingDependentResource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractPollingDependentResource.java
index 2ccba025a7..6355ec39c7 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractPollingDependentResource.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractPollingDependentResource.java
@@ -2,14 +2,13 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.Ignore;
-import io.javaoperatorsdk.operator.processing.dependent.AbstractEventSourceHolderDependentResource;
+import io.javaoperatorsdk.operator.processing.dependent.AbstractExternalDependentResource;
 import io.javaoperatorsdk.operator.processing.event.source.CacheKeyMapper;
 import io.javaoperatorsdk.operator.processing.event.source.ExternalResourceCachingEventSource;
 
 @Ignore
 public abstract class AbstractPollingDependentResource<R, P extends HasMetadata>
-    extends
-    AbstractEventSourceHolderDependentResource<R, P, ExternalResourceCachingEventSource<R, P>>
+    extends AbstractExternalDependentResource<R, P, ExternalResourceCachingEventSource<R, P>>
     implements CacheKeyMapper<R> {
 
   public static final int DEFAULT_POLLING_PERIOD = 5000;
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 070c8f8ffe..e16b2dd924 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
@@ -144,9 +144,10 @@ public Result<R> match(R actualResource, R desired, P primary, Context<P> contex
     return GenericKubernetesResourceMatcher.match(desired, actualResource, false);
   }
 
-  protected void handleDelete(P primary, Context<P> context) {
-    var resource = getSecondaryResource(primary, context);
-    resource.ifPresent(r -> client.resource(r).delete());
+  protected void handleDelete(P primary, R secondary, Context<P> context) {
+    if (secondary != null) {
+      client.resource(secondary).delete();
+    }
   }
 
   @SuppressWarnings("unused")
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/AbstractDependentResourceTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/AbstractDependentResourceTest.java
index b93abd45c0..947596a710 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/AbstractDependentResourceTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/AbstractDependentResourceTest.java
@@ -7,7 +7,6 @@
 import io.fabric8.kubernetes.api.model.ConfigMap;
 import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
-import io.javaoperatorsdk.operator.processing.event.ResourceID;
 import io.javaoperatorsdk.operator.sample.simple.TestCustomResource;
 
 import static org.junit.jupiter.api.Assertions.*;
@@ -85,10 +84,12 @@ public Optional<ConfigMap> getSecondaryResource(TestCustomResource primary,
     }
 
     @Override
-    protected void onCreated(ResourceID primaryResourceId, ConfigMap created) {}
+    protected void onCreated(TestCustomResource primary, ConfigMap created,
+        Context<TestCustomResource> context) {}
 
     @Override
-    protected void onUpdated(ResourceID primaryResourceId, ConfigMap updated, ConfigMap actual) {}
+    protected void onUpdated(TestCustomResource primary, ConfigMap updated, ConfigMap actual,
+        Context<TestCustomResource> context) {}
 
     @Override
     protected ConfigMap desired(TestCustomResource primary, Context<TestCustomResource> context) {
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/EventSourceIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/EventSourceIT.java
index a85543030b..7432e5da0a 100644
--- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/EventSourceIT.java
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/EventSourceIT.java
@@ -16,6 +16,7 @@
 import static org.awaitility.Awaitility.await;
 
 class EventSourceIT {
+
   @RegisterExtension
   LocallyRunOperatorExtension operator =
       LocallyRunOperatorExtension.builder().withReconciler(EventSourceTestCustomReconciler.class)
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/ExternalStateBulkIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/ExternalStateBulkIT.java
new file mode 100644
index 0000000000..a52e9dc3c1
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/ExternalStateBulkIT.java
@@ -0,0 +1,96 @@
+package io.javaoperatorsdk.operator;
+
+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.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension;
+import io.javaoperatorsdk.operator.sample.externalstate.externalstatebulkdependent.ExternalStateBulkDependentCustomResource;
+import io.javaoperatorsdk.operator.sample.externalstate.externalstatebulkdependent.ExternalStateBulkDependentReconciler;
+import io.javaoperatorsdk.operator.sample.externalstate.externalstatebulkdependent.ExternalStateBulkSpec;
+import io.javaoperatorsdk.operator.support.ExternalIDGenServiceMock;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+class ExternalStateBulkIT {
+
+  private static final String TEST_RESOURCE_NAME = "test1";
+
+  public static final String INITIAL_TEST_DATA = "initialTestData";
+  public static final String UPDATED_DATA = "updatedData";
+  public static final int INITIAL_BULK_SIZE = 3;
+  public static final int INCREASED_BULK_SIZE = 4;
+  public static final int DECREASED_BULK_SIZE = 2;
+
+  private ExternalIDGenServiceMock externalService = ExternalIDGenServiceMock.getInstance();
+
+  @RegisterExtension
+  LocallyRunOperatorExtension operator =
+      LocallyRunOperatorExtension.builder()
+          .withReconciler(ExternalStateBulkDependentReconciler.class)
+          .build();
+
+  @Test
+  void reconcilesResourceWithPersistentState() throws InterruptedException {
+    var resource = operator.create(testResource());
+    assertResources(resource, INITIAL_TEST_DATA, INITIAL_BULK_SIZE);
+
+    resource.getSpec().setData(UPDATED_DATA);
+    resource = operator.replace(resource);
+    assertResources(resource, UPDATED_DATA, INITIAL_BULK_SIZE);
+
+    resource.getSpec().setNumber(INCREASED_BULK_SIZE);
+    resource = operator.replace(resource);
+    assertResources(resource, UPDATED_DATA, INCREASED_BULK_SIZE);
+
+    resource.getSpec().setNumber(DECREASED_BULK_SIZE);
+    resource = operator.replace(resource);
+    assertResources(resource, UPDATED_DATA, DECREASED_BULK_SIZE);
+
+    operator.delete(resource);
+    assertResourcesDeleted(resource);
+  }
+
+  private void assertResourcesDeleted(ExternalStateBulkDependentCustomResource resource) {
+    await().untilAsserted(() -> {
+      var configMaps =
+          operator.getKubernetesClient().configMaps().inNamespace(operator.getNamespace())
+              .list().getItems().stream().filter(
+                  cm -> cm.getMetadata().getName().startsWith(resource.getMetadata().getName()));
+      var resources = externalService.listResources();
+      assertThat(configMaps).isEmpty();
+      assertThat(resources).isEmpty();
+    });
+  }
+
+  private void assertResources(ExternalStateBulkDependentCustomResource resource,
+      String initialTestData, int size) {
+    await().pollInterval(Duration.ofMillis(700)).untilAsserted(() -> {
+      var resources = externalService.listResources();
+      assertThat(resources).hasSize(size);
+      assertThat(resources).allMatch(r -> r.getData().startsWith(initialTestData));
+
+      var configMaps =
+          operator.getKubernetesClient().configMaps().inNamespace(operator.getNamespace())
+              .list().getItems().stream().filter(
+                  cm -> cm.getMetadata().getName().startsWith(resource.getMetadata().getName()));
+      assertThat(configMaps).hasSize(size);
+    });
+  }
+
+  private ExternalStateBulkDependentCustomResource testResource() {
+    var res = new ExternalStateBulkDependentCustomResource();
+    res.setMetadata(new ObjectMetaBuilder()
+        .withName(TEST_RESOURCE_NAME)
+        .build());
+
+    res.setSpec(new ExternalStateBulkSpec());
+    res.getSpec().setNumber(INITIAL_BULK_SIZE);
+    res.getSpec().setData(INITIAL_TEST_DATA);
+    return res;
+  }
+
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/ExternalStateDependentIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/ExternalStateDependentIT.java
new file mode 100644
index 0000000000..fd7132bd2a
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/ExternalStateDependentIT.java
@@ -0,0 +1,19 @@
+package io.javaoperatorsdk.operator;
+
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension;
+import io.javaoperatorsdk.operator.sample.externalstate.ExternalStateDependentReconciler;
+
+class ExternalStateDependentIT extends ExternalStateTestBase {
+
+  @RegisterExtension
+  LocallyRunOperatorExtension operator =
+      LocallyRunOperatorExtension.builder().withReconciler(ExternalStateDependentReconciler.class)
+          .build();
+
+  @Override
+  LocallyRunOperatorExtension extension() {
+    return operator;
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/ExternalStateTestBase.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/ExternalStateTestBase.java
new file mode 100644
index 0000000000..972b7e62dc
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/ExternalStateTestBase.java
@@ -0,0 +1,73 @@
+package io.javaoperatorsdk.operator;
+
+import org.junit.jupiter.api.Test;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension;
+import io.javaoperatorsdk.operator.sample.externalstate.ExternalStateCustomResource;
+import io.javaoperatorsdk.operator.sample.externalstate.ExternalStateSpec;
+import io.javaoperatorsdk.operator.support.ExternalIDGenServiceMock;
+
+import static io.javaoperatorsdk.operator.sample.externalstate.ExternalStateReconciler.ID_KEY;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+public abstract class ExternalStateTestBase {
+
+  private static final String TEST_RESOURCE_NAME = "test1";
+
+  public static final String INITIAL_TEST_DATA = "initialTestData";
+  public static final String UPDATED_DATA = "updatedData";
+
+  private ExternalIDGenServiceMock externalService = ExternalIDGenServiceMock.getInstance();
+
+  @Test
+  public void reconcilesResourceWithPersistentState() {
+    var resource = extension().create(testResource());
+    assertResourcesCreated(resource, INITIAL_TEST_DATA);
+
+    resource.getSpec().setData(UPDATED_DATA);
+    extension().replace(resource);
+    assertResourcesCreated(resource, UPDATED_DATA);
+
+    extension().delete(resource);
+    assertResourcesDeleted(resource);
+  }
+
+  private void assertResourcesDeleted(ExternalStateCustomResource resource) {
+    await().untilAsserted(() -> {
+      var cm = extension().get(ConfigMap.class, resource.getMetadata().getName());
+      var resources = externalService.listResources();
+      assertThat(cm).isNull();
+      assertThat(resources).isEmpty();
+    });
+  }
+
+  private void assertResourcesCreated(ExternalStateCustomResource resource,
+      String initialTestData) {
+    await().untilAsserted(() -> {
+      var cm = extension().get(ConfigMap.class, resource.getMetadata().getName());
+      var resources = externalService.listResources();
+      assertThat(resources).hasSize(1);
+      var extRes = externalService.listResources().get(0);
+      assertThat(extRes.getData()).isEqualTo(initialTestData);
+      assertThat(cm).isNotNull();
+      assertThat(cm.getData().get(ID_KEY)).isEqualTo(extRes.getId());
+    });
+  }
+
+  private ExternalStateCustomResource testResource() {
+    var res = new ExternalStateCustomResource();
+    res.setMetadata(new ObjectMetaBuilder()
+        .withName(TEST_RESOURCE_NAME)
+        .build());
+
+    res.setSpec(new ExternalStateSpec());
+    res.getSpec().setData(INITIAL_TEST_DATA);
+    return res;
+  }
+
+  abstract LocallyRunOperatorExtension extension();
+
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/externalstate/ExternalStateDependentReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/externalstate/ExternalStateDependentReconciler.java
new file mode 100644
index 0000000000..8755e7099c
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/externalstate/ExternalStateDependentReconciler.java
@@ -0,0 +1,45 @@
+package io.javaoperatorsdk.operator.sample.externalstate;
+
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.javaoperatorsdk.operator.api.config.informer.InformerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.*;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.Dependent;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
+import io.javaoperatorsdk.operator.support.TestExecutionInfoProvider;
+
+@ControllerConfiguration(
+    dependents = @Dependent(type = ExternalWithStateDependentResource.class))
+public class ExternalStateDependentReconciler
+    implements Reconciler<ExternalStateCustomResource>,
+    EventSourceInitializer<ExternalStateCustomResource>,
+    TestExecutionInfoProvider {
+
+  public static final String ID_KEY = "id";
+  private final AtomicInteger numberOfExecutions = new AtomicInteger(0);
+
+  @Override
+  public UpdateControl<ExternalStateCustomResource> reconcile(
+      ExternalStateCustomResource resource,
+      Context<ExternalStateCustomResource> context) {
+    numberOfExecutions.addAndGet(1);
+
+    return UpdateControl.noUpdate();
+  }
+
+  public int getNumberOfExecutions() {
+    return numberOfExecutions.get();
+  }
+
+  @Override
+  public Map<String, EventSource> prepareEventSources(
+      EventSourceContext<ExternalStateCustomResource> context) {
+    var configMapEventSource = new InformerEventSource<>(
+        InformerConfiguration.from(ConfigMap.class, context).build(), context);
+    return EventSourceInitializer.nameEventSources(configMapEventSource);
+  }
+
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/externalstate/ExternalWithStateDependentResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/externalstate/ExternalWithStateDependentResource.java
new file mode 100644
index 0000000000..b1236b7126
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/externalstate/ExternalWithStateDependentResource.java
@@ -0,0 +1,99 @@
+package io.javaoperatorsdk.operator.sample.externalstate;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.processing.dependent.DependentResourceWithExplicitState;
+import io.javaoperatorsdk.operator.processing.dependent.Matcher;
+import io.javaoperatorsdk.operator.processing.dependent.Updater;
+import io.javaoperatorsdk.operator.processing.dependent.external.PerResourcePollingDependentResource;
+import io.javaoperatorsdk.operator.support.ExternalIDGenServiceMock;
+import io.javaoperatorsdk.operator.support.ExternalResource;
+
+import static io.javaoperatorsdk.operator.sample.externalstate.ExternalStateDependentReconciler.ID_KEY;
+
+public class ExternalWithStateDependentResource extends
+    PerResourcePollingDependentResource<ExternalResource, ExternalStateCustomResource>
+    implements
+    DependentResourceWithExplicitState<ExternalResource, ExternalStateCustomResource, ConfigMap>,
+    Updater<ExternalResource, ExternalStateCustomResource> {
+
+  ExternalIDGenServiceMock externalService = ExternalIDGenServiceMock.getInstance();
+
+  public ExternalWithStateDependentResource() {
+    super(ExternalResource.class, 300);
+  }
+
+  @Override
+  @SuppressWarnings("unchecked")
+  public Set<ExternalResource> fetchResources(
+      ExternalStateCustomResource primaryResource) {
+    Optional<ConfigMap> configMapOptional =
+        getExternalStateEventSource().getSecondaryResource(primaryResource);
+
+    return configMapOptional.map(configMap -> {
+      var id = configMap.getData().get(ID_KEY);
+      var externalResource = externalService.read(id);
+      return externalResource.map(er -> Set.of(er)).orElse(Collections.emptySet());
+    }).orElse(Collections.emptySet());
+  }
+
+  @Override
+  protected ExternalResource desired(ExternalStateCustomResource primary,
+      Context<ExternalStateCustomResource> context) {
+    return new ExternalResource(primary.getSpec().getData());
+  }
+
+  @Override
+  public Class<ConfigMap> stateResourceClass() {
+    return ConfigMap.class;
+  }
+
+  @Override
+  public ConfigMap stateResource(ExternalStateCustomResource primary,
+      ExternalResource resource) {
+    ConfigMap configMap = new ConfigMapBuilder()
+        .withMetadata(new ObjectMetaBuilder()
+            .withName(primary.getMetadata().getName())
+            .withNamespace(primary.getMetadata().getNamespace())
+            .build())
+        .withData(Map.of(ID_KEY, resource.getId()))
+        .build();
+    configMap.addOwnerReference(primary);
+    return configMap;
+  }
+
+  @Override
+  public ExternalResource create(ExternalResource desired,
+      ExternalStateCustomResource primary,
+      Context<ExternalStateCustomResource> context) {
+    return externalService.create(desired);
+  }
+
+  @Override
+  public ExternalResource update(ExternalResource actual,
+      ExternalResource desired, ExternalStateCustomResource primary,
+      Context<ExternalStateCustomResource> context) {
+    return externalService.update(new ExternalResource(actual.getId(), desired.getData()));
+  }
+
+  @Override
+  public Matcher.Result<ExternalResource> match(ExternalResource resource,
+      ExternalStateCustomResource primary,
+      Context<ExternalStateCustomResource> context) {
+    return Matcher.Result.nonComputed(resource.getData().equals(primary.getSpec().getData()));
+  }
+
+  @Override
+  protected void handleDelete(ExternalStateCustomResource primary,
+      ExternalResource secondary,
+      Context<ExternalStateCustomResource> context) {
+    externalService.delete(secondary.getId());
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/externalstate/externalstatebulkdependent/BulkDependentResourceExternalWithState.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/externalstate/externalstatebulkdependent/BulkDependentResourceExternalWithState.java
new file mode 100644
index 0000000000..5344810609
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/externalstate/externalstatebulkdependent/BulkDependentResourceExternalWithState.java
@@ -0,0 +1,137 @@
+package io.javaoperatorsdk.operator.sample.externalstate.externalstatebulkdependent;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.processing.dependent.*;
+import io.javaoperatorsdk.operator.processing.dependent.external.PerResourcePollingDependentResource;
+import io.javaoperatorsdk.operator.support.ExternalIDGenServiceMock;
+import io.javaoperatorsdk.operator.support.ExternalResource;
+
+import static io.javaoperatorsdk.operator.sample.externalstate.ExternalStateDependentReconciler.ID_KEY;
+
+public class BulkDependentResourceExternalWithState extends
+    PerResourcePollingDependentResource<ExternalResource, ExternalStateBulkDependentCustomResource>
+    implements
+    BulkDependentResource<ExternalResource, ExternalStateBulkDependentCustomResource>,
+    DependentResourceWithExplicitState<ExternalResource, ExternalStateBulkDependentCustomResource, ConfigMap>,
+    BulkUpdater<ExternalResource, ExternalStateBulkDependentCustomResource> {
+
+  public static final String DELIMITER = "-";
+  ExternalIDGenServiceMock externalService = ExternalIDGenServiceMock.getInstance();
+
+  public BulkDependentResourceExternalWithState() {
+    super(ExternalResource.class, 300);
+  }
+
+  @Override
+  @SuppressWarnings("unchecked")
+  public Set<ExternalResource> fetchResources(
+      ExternalStateBulkDependentCustomResource primaryResource) {
+    Set<ConfigMap> configMaps =
+        getExternalStateEventSource().getSecondaryResources(primaryResource);
+    Set<ExternalResource> res = new HashSet<>();
+
+    configMaps.stream().forEach(cm -> {
+      var id = cm.getData().get(ID_KEY);
+      var externalResource = externalService.read(id);
+      externalResource.ifPresent(er -> res.add(er));
+    });
+    return res;
+  }
+
+  @Override
+  public Class<ConfigMap> stateResourceClass() {
+    return ConfigMap.class;
+  }
+
+  @Override
+  public ConfigMap stateResource(ExternalStateBulkDependentCustomResource primary,
+      ExternalResource resource) {
+    ConfigMap configMap = new ConfigMapBuilder()
+        .withMetadata(new ObjectMetaBuilder()
+            .withName(configMapName(primary, resource))
+            .withNamespace(primary.getMetadata().getNamespace())
+            .build())
+        .withData(Map.of(ID_KEY, resource.getId()))
+        .build();
+    configMap.addOwnerReference(primary);
+    return configMap;
+  }
+
+  @Override
+  public ExternalResource create(ExternalResource desired,
+      ExternalStateBulkDependentCustomResource primary,
+      Context<ExternalStateBulkDependentCustomResource> context) {
+    return externalService.create(desired);
+  }
+
+  @Override
+  public ExternalResource update(ExternalResource actual,
+      ExternalResource desired, ExternalStateBulkDependentCustomResource primary,
+      Context<ExternalStateBulkDependentCustomResource> context) {
+    return externalService.update(new ExternalResource(actual.getId(), desired.getData()));
+  }
+
+  @Override
+  protected void handleDelete(ExternalStateBulkDependentCustomResource primary,
+      ExternalResource secondary,
+      Context<ExternalStateBulkDependentCustomResource> context) {
+    externalService.delete(secondary.getId());
+  }
+
+  @Override
+  public Matcher.Result<ExternalResource> match(ExternalResource actualResource,
+      ExternalResource desired,
+      ExternalStateBulkDependentCustomResource primary,
+      Context<ExternalStateBulkDependentCustomResource> context) {
+    return Matcher.Result.computed(desired.getData().equals(actualResource.getData()), desired);
+  }
+
+  @Override
+  public Map<String, ExternalResource> desiredResources(
+      ExternalStateBulkDependentCustomResource primary,
+      Context<ExternalStateBulkDependentCustomResource> context) {
+    int number = primary.getSpec().getNumber();
+    Map<String, ExternalResource> res = new HashMap<>();
+    for (int i = 0; i < number; i++) {
+      res.put(Integer.toString(i),
+          new ExternalResource(primary.getSpec().getData() + DELIMITER + i));
+    }
+    return res;
+  }
+
+  @Override
+  public Map<String, ExternalResource> getSecondaryResources(
+      ExternalStateBulkDependentCustomResource primary,
+      Context<ExternalStateBulkDependentCustomResource> context) {
+    var resources = context.getSecondaryResources(ExternalResource.class);
+    return resources.stream().collect(Collectors.toMap(this::externalResourceIndex, r -> r));
+  }
+
+  @Override
+  public void handleDeleteTargetResource(ExternalStateBulkDependentCustomResource primary,
+      ExternalResource resource, String key,
+      Context<ExternalStateBulkDependentCustomResource> context) {
+    externalService.delete(resource.getId());
+  }
+
+  private String externalResourceIndex(ExternalResource externalResource) {
+    return externalResource.getData()
+        .substring(externalResource.getData().lastIndexOf(DELIMITER) + 1);
+  }
+
+  private String configMapName(ExternalStateBulkDependentCustomResource primary,
+      ExternalResource resource) {
+    return primary.getMetadata().getName() + DELIMITER + externalResourceIndex(resource);
+  }
+
+  @Override
+  public String keyFor(ExternalResource resource) {
+    return externalResourceIndex(resource);
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/externalstate/externalstatebulkdependent/ExternalStateBulkDependentCustomResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/externalstate/externalstatebulkdependent/ExternalStateBulkDependentCustomResource.java
new file mode 100644
index 0000000000..9ee3e260e8
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/externalstate/externalstatebulkdependent/ExternalStateBulkDependentCustomResource.java
@@ -0,0 +1,15 @@
+package io.javaoperatorsdk.operator.sample.externalstate.externalstatebulkdependent;
+
+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("esb")
+public class ExternalStateBulkDependentCustomResource
+    extends CustomResource<ExternalStateBulkSpec, Void>
+    implements Namespaced {
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/externalstate/externalstatebulkdependent/ExternalStateBulkDependentReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/externalstate/externalstatebulkdependent/ExternalStateBulkDependentReconciler.java
new file mode 100644
index 0000000000..a572572a97
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/externalstate/externalstatebulkdependent/ExternalStateBulkDependentReconciler.java
@@ -0,0 +1,46 @@
+package io.javaoperatorsdk.operator.sample.externalstate.externalstatebulkdependent;
+
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.javaoperatorsdk.operator.api.config.informer.InformerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.*;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.Dependent;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
+import io.javaoperatorsdk.operator.support.TestExecutionInfoProvider;
+
+@ControllerConfiguration(
+    dependents = @Dependent(
+        type = BulkDependentResourceExternalWithState.class))
+public class ExternalStateBulkDependentReconciler
+    implements Reconciler<ExternalStateBulkDependentCustomResource>,
+    EventSourceInitializer<ExternalStateBulkDependentCustomResource>,
+    TestExecutionInfoProvider {
+
+  public static final String ID_KEY = "id";
+  private final AtomicInteger numberOfExecutions = new AtomicInteger(0);
+
+  @Override
+  public UpdateControl<ExternalStateBulkDependentCustomResource> reconcile(
+      ExternalStateBulkDependentCustomResource resource,
+      Context<ExternalStateBulkDependentCustomResource> context) {
+    numberOfExecutions.addAndGet(1);
+
+    return UpdateControl.noUpdate();
+  }
+
+  public int getNumberOfExecutions() {
+    return numberOfExecutions.get();
+  }
+
+  @Override
+  public Map<String, EventSource> prepareEventSources(
+      EventSourceContext<ExternalStateBulkDependentCustomResource> context) {
+    var configMapEventSource = new InformerEventSource<>(
+        InformerConfiguration.from(ConfigMap.class, context).build(), context);
+    return EventSourceInitializer.nameEventSources(configMapEventSource);
+  }
+
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/externalstate/externalstatebulkdependent/ExternalStateBulkSpec.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/externalstate/externalstatebulkdependent/ExternalStateBulkSpec.java
new file mode 100644
index 0000000000..ad52a389fd
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/externalstate/externalstatebulkdependent/ExternalStateBulkSpec.java
@@ -0,0 +1,25 @@
+package io.javaoperatorsdk.operator.sample.externalstate.externalstatebulkdependent;
+
+public class ExternalStateBulkSpec {
+
+  private Integer number;
+  private String data;
+
+  public String getData() {
+    return data;
+  }
+
+  public ExternalStateBulkSpec setData(String data) {
+    this.data = data;
+    return this;
+  }
+
+  public Integer getNumber() {
+    return number;
+  }
+
+  public ExternalStateBulkSpec setNumber(Integer number) {
+    this.number = number;
+    return this;
+  }
+}