diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ResourceDiscriminator.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ResourceDiscriminator.java
index 072e7d8078..eb947fa440 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ResourceDiscriminator.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ResourceDiscriminator.java
@@ -7,5 +7,4 @@
 public interface ResourceDiscriminator<R, P extends HasMetadata> {
 
   Optional<R> distinguish(Class<R> resource, P primary, Context<P> context);
-
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/ReconcileResult.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/ReconcileResult.java
index c83da1c8ea..468e14e8ea 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/ReconcileResult.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/ReconcileResult.java
@@ -1,14 +1,14 @@
 package io.javaoperatorsdk.operator.api.reconciler.dependent;
 
-import java.util.Optional;
+import java.util.*;
+import java.util.stream.Collectors;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 
 public class ReconcileResult<R> {
 
-  private final R resource;
-  private final Operation operation;
+  private final Map<R, Operation> resourceOperations;
 
   public static <T> ReconcileResult<T> resourceCreated(T resource) {
     return new ReconcileResult<>(resource, Operation.CREATED);
@@ -22,25 +22,49 @@ public static <T> ReconcileResult<T> noOperation(T resource) {
     return new ReconcileResult<>(resource, Operation.NONE);
   }
 
+  @SafeVarargs
+  public static <T> ReconcileResult<T> aggregatedResult(ReconcileResult<T>... results) {
+    if (results == null) {
+      throw new IllegalArgumentException("Should provide results to aggregate");
+    }
+    if (results.length == 1) {
+      return results[0];
+    }
+    final Map<T, Operation> operations = new HashMap<>(results.length);
+    for (ReconcileResult<T> res : results) {
+      res.getSingleResource().ifPresent(r -> operations.put(r, res.getSingleOperation()));
+    }
+    return new ReconcileResult<>(operations);
+  }
+
   @Override
   public String toString() {
-    return getResource()
-        .map(r -> r instanceof HasMetadata ? ResourceID.fromResource((HasMetadata) r) : r)
-        .orElse("no resource")
-        + " -> " + operation;
+    return resourceOperations.entrySet().stream().collect(Collectors.toMap(
+        e -> e instanceof HasMetadata ? ResourceID.fromResource((HasMetadata) e) : e,
+        Map.Entry::getValue))
+        .toString();
   }
 
   private ReconcileResult(R resource, Operation operation) {
-    this.resource = resource;
-    this.operation = operation;
+    resourceOperations = resource != null ? Map.of(resource, operation) : Collections.emptyMap();
+  }
+
+  private ReconcileResult(Map<R, Operation> operations) {
+    resourceOperations = Collections.unmodifiableMap(operations);
+  }
+
+  public Optional<R> getSingleResource() {
+    return resourceOperations.entrySet().stream().findFirst().map(Map.Entry::getKey);
   }
 
-  public Optional<R> getResource() {
-    return Optional.ofNullable(resource);
+  public Operation getSingleOperation() {
+    return resourceOperations.entrySet().stream().findFirst().map(Map.Entry::getValue)
+        .orElseThrow();
   }
 
-  public Operation getOperation() {
-    return operation;
+  @SuppressWarnings("unused")
+  public Map<R, Operation> getResourceOperations() {
+    return resourceOperations;
   }
 
   public enum Operation {
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 1abfb3df4b..078fc60b66 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractDependentResource.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractDependentResource.java
@@ -1,5 +1,7 @@
 package io.javaoperatorsdk.operator.processing.dependent;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Optional;
 
 import org.slf4j.Logger;
@@ -20,25 +22,73 @@ public abstract class AbstractDependentResource<R, P extends HasMetadata>
 
   protected final boolean creatable = this instanceof Creator;
   protected final boolean updatable = this instanceof Updater;
+  protected final boolean bulk = this instanceof BulkDependentResource;
 
   protected Creator<R, P> creator;
   protected Updater<R, P> updater;
+  protected BulkDependentResource<R, P> bulkDependentResource;
 
-  private ResourceDiscriminator<R, P> resourceDiscriminator;
+  private final List<ResourceDiscriminator<R, P>> resourceDiscriminator = new ArrayList<>(1);
 
   @SuppressWarnings("unchecked")
   public AbstractDependentResource() {
     creator = creatable ? (Creator<R, P>) this : null;
     updater = updatable ? (Updater<R, P>) this : null;
+
+    bulkDependentResource = bulk ? (BulkDependentResource<R, P>) this : null;
   }
 
   @Override
   public ReconcileResult<R> reconcile(P primary, Context<P> context) {
-    Optional<R> maybeActual = getSecondaryResource(primary, context);
+    if (bulk) {
+      final var count = bulkDependentResource.count(primary, context);
+      deleteBulkResourcesIfRequired(count, lastKnownBulkSize(), primary, context);
+      adjustDiscriminators(count);
+      @SuppressWarnings("unchecked")
+      final ReconcileResult<R>[] results = new ReconcileResult[count];
+      for (int i = 0; i < count; i++) {
+        results[i] = reconcileIndexAware(primary, i, context);
+      }
+      return ReconcileResult.aggregatedResult(results);
+    } else {
+      return reconcileIndexAware(primary, 0, context);
+    }
+  }
+
+  protected void deleteBulkResourcesIfRequired(int targetCount, int actualCount, P primary,
+      Context<P> context) {
+    if (targetCount >= actualCount) {
+      return;
+    }
+    for (int i = targetCount; i < actualCount; i++) {
+      var resource = getSecondaryResourceIndexAware(primary, i, context);
+      var index = i;
+      resource.ifPresent(
+          r -> bulkDependentResource.deleteBulkResourceWithIndex(primary, r, index, context));
+    }
+  }
+
+  private void adjustDiscriminators(int count) {
+    if (resourceDiscriminator.size() == count) {
+      return;
+    }
+    if (resourceDiscriminator.size() < count) {
+      for (int i = resourceDiscriminator.size(); i < count; i++) {
+        resourceDiscriminator.add(bulkDependentResource.getResourceDiscriminator(i));
+      }
+    }
+    if (resourceDiscriminator.size() > count) {
+      resourceDiscriminator.subList(count, resourceDiscriminator.size()).clear();
+    }
+  }
+
+  protected ReconcileResult<R> reconcileIndexAware(P primary, int i, Context<P> context) {
+    Optional<R> maybeActual = bulk ? getSecondaryResourceIndexAware(primary, i, context)
+        : getSecondaryResource(primary, context);
     if (creatable || updatable) {
       if (maybeActual.isEmpty()) {
         if (creatable) {
-          var desired = desired(primary, context);
+          var desired = desiredIndexAware(primary, i, context);
           throwIfNull(desired, primary, "Desired");
           logForOperation("Creating", primary, desired);
           var createdResource = handleCreate(desired, primary, context);
@@ -47,9 +97,15 @@ public ReconcileResult<R> reconcile(P primary, Context<P> context) {
       } else {
         final var actual = maybeActual.get();
         if (updatable) {
-          final var match = updater.match(actual, primary, context);
+          final Matcher.Result<R> match;
+          if (bulk) {
+            match = updater.match(actual, primary, i, context);
+          } else {
+            match = updater.match(actual, primary, context);
+          }
           if (!match.matched()) {
-            final var desired = match.computedDesired().orElse(desired(primary, context));
+            final var desired =
+                match.computedDesired().orElse(desiredIndexAware(primary, i, context));
             throwIfNull(desired, primary, "Desired");
             logForOperation("Updating", primary, desired);
             var updatedResource = handleUpdate(actual, desired, primary, context);
@@ -67,9 +123,18 @@ public ReconcileResult<R> reconcile(P primary, Context<P> context) {
     return ReconcileResult.noOperation(maybeActual.orElse(null));
   }
 
+  private R desiredIndexAware(P primary, int i, Context<P> context) {
+    return bulk ? desired(primary, i, context)
+        : desired(primary, context);
+  }
+
   public Optional<R> getSecondaryResource(P primary, Context<P> context) {
-    return resourceDiscriminator == null ? context.getSecondaryResource(resourceType())
-        : resourceDiscriminator.distinguish(resourceType(), primary, context);
+    return resourceDiscriminator.isEmpty() ? context.getSecondaryResource(resourceType())
+        : resourceDiscriminator.get(0).distinguish(resourceType(), primary, context);
+  }
+
+  protected Optional<R> getSecondaryResourceIndexAware(P primary, int index, Context<P> context) {
+    return context.getSecondaryResource(resourceType(), resourceDiscriminator.get(index));
   }
 
   private void throwIfNull(R desired, P primary, String descriptor) {
@@ -97,7 +162,7 @@ protected R handleCreate(R desired, P primary, Context<P> context) {
   }
 
   /**
-   * Allows sub-classes to perform additional processing (e.g. caching) on the created resource if
+   * 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
@@ -129,12 +194,29 @@ protected R desired(P primary, Context<P> context) {
         "desired method must be implemented if this DependentResource can be created and/or updated");
   }
 
-  public void setResourceDiscriminator(
+  protected R desired(P primary, int index, Context<P> context) {
+    throw new IllegalStateException(
+        "Must be implemented for bulk DependentResource creation");
+  }
+
+  public AbstractDependentResource<R, P> setResourceDiscriminator(
       ResourceDiscriminator<R, P> resourceDiscriminator) {
-    this.resourceDiscriminator = resourceDiscriminator;
+    if (resourceDiscriminator != null) {
+      this.resourceDiscriminator.add(resourceDiscriminator);
+    }
+    return this;
   }
 
   public ResourceDiscriminator<R, P> getResourceDiscriminator() {
-    return resourceDiscriminator;
+    if (this.resourceDiscriminator.isEmpty()) {
+      return null;
+    } else {
+      return this.resourceDiscriminator.get(0);
+    }
   }
+
+  protected int lastKnownBulkSize() {
+    return resourceDiscriminator.size();
+  }
+
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/BulkDependentResource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/BulkDependentResource.java
new file mode 100644
index 0000000000..1f2688f5cb
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/BulkDependentResource.java
@@ -0,0 +1,35 @@
+package io.javaoperatorsdk.operator.processing.dependent;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ResourceDiscriminator;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.Deleter;
+
+/**
+ * Manages dynamic number of resources created for a primary resource. Since the point of a bulk
+ * dependent resource is to manage the number of secondary resources dynamically it implement
+ * {@link Creator} and {@link Deleter} interfaces out of the box. A concrete dependent resource can
+ * implement additionally also {@link Updater}.
+ */
+public interface BulkDependentResource<R, P extends HasMetadata> extends Creator<R, P>, Deleter<P> {
+
+  /**
+   * @return number of resources to create
+   */
+  int count(P primary, Context<P> context);
+
+  R desired(P primary, int index, Context<P> context);
+
+  /**
+   * Used to delete resource if the desired count is lower than the actual count of a resource.
+   *
+   * @param primary resource
+   * @param resource actual resource from the cache for the index
+   * @param i index of the resource
+   * @param context actual context
+   */
+  void deleteBulkResourceWithIndex(P primary, R resource, int i, Context<P> context);
+
+  ResourceDiscriminator<R, P> getResourceDiscriminator(int index);
+
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/BulkUpdater.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/BulkUpdater.java
new file mode 100644
index 0000000000..9c00b47d0c
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/BulkUpdater.java
@@ -0,0 +1,20 @@
+package io.javaoperatorsdk.operator.processing.dependent;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+
+/**
+ * Helper for the Bulk Dependent Resources to make it more explicit that bulk needs to only
+ * implement the index aware match method.
+ *
+ * @param <R> secondary resource type
+ * @param <P> primary resource type
+ */
+public interface BulkUpdater<R, P extends HasMetadata> extends Updater<R, P> {
+
+  default Matcher.Result<R> match(R actualResource, P primary, Context<P> context) {
+    throw new IllegalStateException();
+  }
+
+  Matcher.Result<R> match(R actualResource, P primary, int index, Context<P> context);
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/DesiredEqualsMatcher.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/DesiredEqualsMatcher.java
index 459d7951d6..1d3b34a47b 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/DesiredEqualsMatcher.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/DesiredEqualsMatcher.java
@@ -16,4 +16,10 @@ public Result<R> match(R actualResource, P primary, Context<P> context) {
     var desired = abstractDependentResource.desired(primary, context);
     return Result.computed(actualResource.equals(desired), desired);
   }
+
+  @Override
+  public Result<R> match(R actualResource, P primary, int index, Context<P> context) {
+    var desired = abstractDependentResource.desired(primary, index, context);
+    return Result.computed(actualResource.equals(desired), desired);
+  }
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/Matcher.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/Matcher.java
index 750fe89cbf..835f76ab3a 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/Matcher.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/Matcher.java
@@ -95,4 +95,19 @@ public Optional<T> computedDesired() {
    *         {@link Result#computed(boolean, Object)})
    */
   Result<R> match(R actualResource, P primary, Context<P> context);
+
+  /**
+   * Determines whether the specified secondary resource matches the desired state with target index
+   * of a bulk resource as defined from the specified primary resource, given the specified
+   * {@link Context}.
+   *
+   * @param actualResource the resource we want to determine whether it's matching the desired state
+   * @param primary the primary resource from which the desired state is inferred
+   * @param context the context in which the resource is being matched
+   * @return a {@link Result} encapsulating whether the resource matched its desired state and this
+   *         associated state if it was computed as part of the matching process. Use the static
+   *         convenience methods ({@link Result#nonComputed(boolean)} and
+   *         {@link Result#computed(boolean, Object)})
+   */
+  Result<R> match(R actualResource, P primary, int index, Context<P> context);
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/Updater.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/Updater.java
index 828f9ad785..06b3cb52f6 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/Updater.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/Updater.java
@@ -8,4 +8,8 @@ public interface Updater<R, P extends HasMetadata> {
   R update(R actual, R desired, P primary, Context<P> context);
 
   Result<R> match(R actualResource, P primary, Context<P> context);
+
+  default Result<R> match(R actualResource, P primary, int index, Context<P> context) {
+    throw new IllegalStateException("Implement this for bulk matching");
+  }
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/GenericKubernetesResourceMatcher.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/GenericKubernetesResourceMatcher.java
index e294b1c938..bb066b5b24 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/GenericKubernetesResourceMatcher.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/GenericKubernetesResourceMatcher.java
@@ -24,17 +24,42 @@ private GenericKubernetesResourceMatcher(KubernetesDependentResource<R, P> depen
   static <R extends HasMetadata, P extends HasMetadata> Matcher<R, P> matcherFor(
       Class<R> resourceType, KubernetesDependentResource<R, P> dependentResource) {
     if (Secret.class.isAssignableFrom(resourceType)) {
-      return (actual, primary, context) -> {
-        final var desired = dependentResource.desired(primary, context);
-        return Result.computed(
-            ResourceComparators.compareSecretData((Secret) desired, (Secret) actual), desired);
+      return new Matcher<>() {
+        @Override
+        public Result<R> match(R actualResource, P primary, Context<P> context) {
+          final var desired = dependentResource.desired(primary, context);
+          return Result.computed(
+              ResourceComparators.compareSecretData((Secret) desired, (Secret) actualResource),
+              desired);
+        }
+
+        @Override
+        public Result<R> match(R actualResource, P primary, int index, Context<P> context) {
+          final var desired = dependentResource.desired(primary, index, context);
+          return Result.computed(
+              ResourceComparators.compareSecretData((Secret) desired, (Secret) actualResource),
+              desired);
+        }
       };
     } else if (ConfigMap.class.isAssignableFrom(resourceType)) {
-      return (actual, primary, context) -> {
-        final var desired = dependentResource.desired(primary, context);
-        return Result.computed(
-            ResourceComparators.compareConfigMapData((ConfigMap) desired, (ConfigMap) actual),
-            desired);
+      return new Matcher<>() {
+        @Override
+        public Result<R> match(R actualResource, P primary, Context<P> context) {
+          final var desired = dependentResource.desired(primary, context);
+          return Result.computed(
+              ResourceComparators.compareConfigMapData((ConfigMap) desired,
+                  (ConfigMap) actualResource),
+              desired);
+        }
+
+        @Override
+        public Result<R> match(R actualResource, P primary, int index, Context<P> context) {
+          final var desired = dependentResource.desired(primary, index, context);
+          return Result.computed(
+              ResourceComparators.compareConfigMapData((ConfigMap) desired,
+                  (ConfigMap) actualResource),
+              desired);
+        }
       };
     } else {
       return new GenericKubernetesResourceMatcher(dependentResource);
@@ -43,32 +68,18 @@ static <R extends HasMetadata, P extends HasMetadata> Matcher<R, P> matcherFor(
 
   @Override
   public Result<R> match(R actualResource, P primary, Context<P> context) {
-    return match(dependentResource, actualResource, primary, context, false);
+    var desired = dependentResource.desired(primary, context);
+    return match(desired, actualResource, false);
   }
 
-  /**
-   * Determines whether the specified actual resource matches the desired state defined by the
-   * specified {@link KubernetesDependentResource} based on the observed state of the associated
-   * specified primary resource.
-   *
-   * @param dependentResource the {@link KubernetesDependentResource} implementation used to
-   *        computed the desired state associated with the specified primary resource
-   * @param actualResource the observed dependent resource for which we want to determine whether it
-   *        matches the desired state or not
-   * @param primary the primary resource from which we want to compute the desired state
-   * @param context the {@link Context} instance within which this method is called
-   * @param considerMetadata {@code true} to consider the metadata of the actual resource when
-   *        determining if it matches the desired state, {@code false} if matching should occur only
-   *        considering the spec of the resources
-   * @return a {@link io.javaoperatorsdk.operator.processing.dependent.Matcher.Result} object
-   * @param <R> the type of resource we want to determine whether they match or not
-   * @param <P> the type of primary resources associated with the secondary resources we want to
-   *        match
-   */
-  public static <R extends HasMetadata, P extends HasMetadata> Result<R> match(
-      KubernetesDependentResource<R, P> dependentResource, R actualResource, P primary,
-      Context<P> context, boolean considerMetadata) {
-    final var desired = dependentResource.desired(primary, context);
+  @Override
+  public Result<R> match(R actualResource, P primary, int index, Context<P> context) {
+    var desired = dependentResource.desired(primary, index, context);
+    return match(desired, actualResource, false);
+  }
+
+  public static <R extends HasMetadata> Result<R> match(
+      R desired, R actualResource, boolean considerMetadata) {
     if (considerMetadata) {
       final var desiredMetadata = desired.getMetadata();
       final var actualMetadata = actualResource.getMetadata();
@@ -95,4 +106,30 @@ public static <R extends HasMetadata, P extends HasMetadata> Result<R> match(
     }
     return Result.computed(true, desired);
   }
+
+  /**
+   * Determines whether the specified actual resource matches the desired state defined by the
+   * specified {@link KubernetesDependentResource} based on the observed state of the associated
+   * specified primary resource.
+   *
+   * @param dependentResource the {@link KubernetesDependentResource} implementation used to
+   *        computed the desired state associated with the specified primary resource
+   * @param actualResource the observed dependent resource for which we want to determine whether it
+   *        matches the desired state or not
+   * @param primary the primary resource from which we want to compute the desired state
+   * @param context the {@link Context} instance within which this method is called
+   * @param considerMetadata {@code true} to consider the metadata of the actual resource when
+   *        determining if it matches the desired state, {@code false} if matching should occur only
+   *        considering the spec of the resources
+   * @return a {@link io.javaoperatorsdk.operator.processing.dependent.Matcher.Result} object
+   * @param <R> the type of resource we want to determine whether they match or not
+   * @param <P> the type of primary resources associated with the secondary resources we want to
+   *        match
+   */
+  public static <R extends HasMetadata, P extends HasMetadata> Result<R> match(
+      KubernetesDependentResource<R, P> dependentResource, R actualResource, P primary,
+      Context<P> context, boolean considerMetadata) {
+    final var desired = dependentResource.desired(primary, context);
+    return match(desired, actualResource, considerMetadata);
+  }
 }
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 328a061e6b..3738a2e7d2 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
@@ -134,8 +134,21 @@ public Result<R> match(R actualResource, P primary, Context<P> context) {
     return matcher.match(actualResource, primary, context);
   }
 
+  public Result<R> match(R actualResource, P primary, int index, Context<P> context) {
+    return matcher.match(actualResource, primary, index, context);
+  }
+
   public void delete(P primary, Context<P> context) {
-    getSecondaryResource(primary, context).ifPresent(r -> client.resource(r).delete());
+    if (bulk) {
+      deleteBulkResourcesIfRequired(0, lastKnownBulkSize(), primary, context);
+    } else {
+      var resource = getSecondaryResource(primary, context);
+      resource.ifPresent(r -> client.resource(r).delete());
+    }
+  }
+
+  public void deleteBulkResourceWithIndex(P primary, R resource, int i, Context<P> context) {
+    client.resource(resource).delete();
   }
 
   @SuppressWarnings("unchecked")
@@ -149,9 +162,7 @@ protected Resource<R> prepare(R desired, P primary, String actionName) {
     } else if (useDefaultAnnotationsToIdentifyPrimary()) {
       addDefaultSecondaryToPrimaryMapperAnnotations(desired, primary);
     }
-    Class<R> targetClass = (Class<R>) desired.getClass();
-    return client.resources(targetClass).inNamespace(desired.getMetadata().getNamespace())
-        .resource(desired);
+    return client.resource(desired).inNamespace(desired.getMetadata().getNamespace());
   }
 
   @Override
@@ -163,8 +174,10 @@ protected InformerEventSource<R, P> createEventSource(EventSourceContext<P> cont
       onUpdateFilter = kubernetesDependentResourceConfig.onUpdateFilter();
       onDeleteFilter = kubernetesDependentResourceConfig.onDeleteFilter();
       genericFilter = kubernetesDependentResourceConfig.genericFilter();
-      setResourceDiscriminator(kubernetesDependentResourceConfig.getResourceDiscriminator());
-
+      var discriminator = kubernetesDependentResourceConfig.getResourceDiscriminator();
+      if (discriminator != null) {
+        setResourceDiscriminator(discriminator);
+      }
       configureWith(kubernetesDependentResourceConfig.labelSelector(),
           kubernetesDependentResourceConfig.namespaces(),
           !kubernetesDependentResourceConfig.wereNamespacesConfigured(), context);
@@ -215,6 +228,11 @@ protected R desired(P primary, Context<P> context) {
     return super.desired(primary, context);
   }
 
+  @Override
+  protected R desired(P primary, int index, Context<P> context) {
+    return super.desired(primary, index, context);
+  }
+
   private void prepareEventFiltering(R desired, ResourceID resourceID) {
     eventSource().prepareForCreateOrUpdateEventFiltering(resourceID, desired);
   }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResourceConfig.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResourceConfig.java
index e2a2c0f684..c7674dd1a7 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResourceConfig.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResourceConfig.java
@@ -75,7 +75,6 @@ public OnAddFilter onAddFilter() {
     return onAddFilter;
   }
 
-
   public OnUpdateFilter<R> onUpdateFilter() {
     return onUpdateFilter;
   }
@@ -92,4 +91,10 @@ public GenericFilter<R> genericFilter() {
   public ResourceDiscriminator getResourceDiscriminator() {
     return resourceDiscriminator;
   }
+
+  public <P> KubernetesDependentResourceConfig<R> setResourceDiscriminator(
+      ResourceDiscriminator<R, ?> resourceDiscriminator) {
+    this.resourceDiscriminator = resourceDiscriminator;
+    return this;
+  }
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowCleanupExecutor.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowCleanupExecutor.java
index 45541b91d6..ac08d2d874 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowCleanupExecutor.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowCleanupExecutor.java
@@ -125,7 +125,6 @@ public void run() {
     }
   }
 
-
   private synchronized void handleDependentCleaned(
       DependentResourceNode<?, P> dependentResourceNode) {
     var dependOns = dependentResourceNode.getDependsOn();
diff --git a/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/LocallyRunOperatorExtension.java b/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/LocallyRunOperatorExtension.java
index e2f4234453..f7627a5555 100644
--- a/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/LocallyRunOperatorExtension.java
+++ b/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/LocallyRunOperatorExtension.java
@@ -134,7 +134,10 @@ protected void before(ExtensionContext context) {
         ref.controllerConfigurationOverrider.accept(oconfig);
       }
 
-      applyCrd(config.getResourceTypeName());
+      // only try to apply a CRD for the reconciler if it is associated to a CR
+      if (CustomResource.class.isAssignableFrom(config.getResourceClass())) {
+        applyCrd(config.getResourceTypeName());
+      }
 
       if (ref.reconciler instanceof KubernetesClientAware) {
         ((KubernetesClientAware) ref.reconciler).setKubernetesClient(kubernetesClient);
@@ -151,6 +154,9 @@ protected void before(ExtensionContext context) {
   private void applyCrd(String resourceTypeName) {
     String path = "/META-INF/fabric8/" + resourceTypeName + "-v1.yml";
     try (InputStream is = getClass().getResourceAsStream(path)) {
+      if (is == null) {
+        throw new IllegalStateException("Cannot find CRD at " + path);
+      }
       final var crd = getKubernetesClient().load(is);
       crd.createOrReplace();
       Thread.sleep(CRD_READY_WAIT); // readiness is not applicable for CRD, just wait a little
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/bulkdependent/BulkDependentDeleterIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/bulkdependent/BulkDependentDeleterIT.java
new file mode 100644
index 0000000000..a934bdd1f3
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/bulkdependent/BulkDependentDeleterIT.java
@@ -0,0 +1,19 @@
+package io.javaoperatorsdk.operator.bulkdependent;
+
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension;
+import io.javaoperatorsdk.operator.sample.bulkdependent.ManagedDeleterBulkReconciler;
+
+public class BulkDependentDeleterIT extends BulkDependentTestBase {
+
+  @RegisterExtension
+  LocallyRunOperatorExtension extension =
+      LocallyRunOperatorExtension.builder().withReconciler(new ManagedDeleterBulkReconciler())
+          .build();
+
+  @Override
+  LocallyRunOperatorExtension extension() {
+    return extension;
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/bulkdependent/BulkDependentTestBase.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/bulkdependent/BulkDependentTestBase.java
new file mode 100644
index 0000000000..605731623c
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/bulkdependent/BulkDependentTestBase.java
@@ -0,0 +1,114 @@
+package io.javaoperatorsdk.operator.bulkdependent;
+
+import java.time.Duration;
+
+import org.junit.jupiter.api.Test;
+
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension;
+import io.javaoperatorsdk.operator.sample.bulkdependent.BulkDependentTestCustomResource;
+import io.javaoperatorsdk.operator.sample.bulkdependent.BulkDependentTestSpec;
+import io.javaoperatorsdk.operator.sample.bulkdependent.ConfigMapDeleterBulkDependentResource;
+
+import static io.javaoperatorsdk.operator.sample.bulkdependent.ConfigMapDeleterBulkDependentResource.LABEL_KEY;
+import static io.javaoperatorsdk.operator.sample.bulkdependent.ConfigMapDeleterBulkDependentResource.LABEL_VALUE;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+public abstract class BulkDependentTestBase {
+
+  public static final String TEST_RESOURCE_NAME = "test";
+  public static final int INITIAL_NUMBER_OF_CONFIG_MAPS = 3;
+  public static final String INITIAL_ADDITIONAL_DATA = "initialData";
+  public static final String NEW_VERSION_OF_ADDITIONAL_DATA = "newVersionOfAdditionalData";
+
+  @Test
+  public void managesBulkConfigMaps() {
+    extension().create(testResource());
+    assertNumberOfConfigMaps(3);
+
+    updateSpecWithNumber(1);
+    assertNumberOfConfigMaps(1);
+
+    updateSpecWithNumber(5);
+    assertNumberOfConfigMaps(5);
+
+    extension().delete(testResource());
+    assertNumberOfConfigMaps(0);
+  }
+
+  @Test
+  public void updatesData() {
+    extension().create(testResource());
+    assertNumberOfConfigMaps(3);
+    assertAdditionalDataOnConfigMaps(INITIAL_ADDITIONAL_DATA);
+
+    updateSpecWithNewAdditionalData(NEW_VERSION_OF_ADDITIONAL_DATA);
+    assertAdditionalDataOnConfigMaps(NEW_VERSION_OF_ADDITIONAL_DATA);
+  }
+
+  private void assertNumberOfConfigMaps(int n) {
+    // this test was failing with a lower timeout on GitHub, probably the garbage collection was
+    // slower there.
+    await().atMost(Duration.ofSeconds(30))
+        .untilAsserted(() -> {
+          var cms =
+              extension().getKubernetesClient().configMaps().inNamespace(extension().getNamespace())
+                  .withLabel(LABEL_KEY, LABEL_VALUE)
+                  .list().getItems();
+          assertThat(cms).withFailMessage("Number of items is still: " + cms.size())
+              .hasSize(n);
+        });
+  }
+
+  private void assertAdditionalDataOnConfigMaps(String expectedValue) {
+    await().atMost(Duration.ofSeconds(30))
+        .untilAsserted(() -> {
+          var cms =
+              extension().getKubernetesClient().configMaps().inNamespace(extension().getNamespace())
+                  .withLabel(LABEL_KEY, LABEL_VALUE)
+                  .list().getItems();
+          cms.forEach(cm -> {
+            assertThat(cm.getData().get(ConfigMapDeleterBulkDependentResource.ADDITIONAL_DATA_KEY))
+                .isEqualTo(expectedValue);
+          });
+        });
+  }
+
+  public static BulkDependentTestCustomResource testResource() {
+    BulkDependentTestCustomResource cr = new BulkDependentTestCustomResource();
+    cr.setMetadata(new ObjectMeta());
+    cr.getMetadata().setName(TEST_RESOURCE_NAME);
+    cr.setSpec(new BulkDependentTestSpec());
+    cr.getSpec().setNumberOfResources(INITIAL_NUMBER_OF_CONFIG_MAPS);
+    cr.getSpec().setAdditionalData(INITIAL_ADDITIONAL_DATA);
+    return cr;
+  }
+
+  private void updateSpecWithNewAdditionalData(String data) {
+    var resource = testResource();
+    resource.getSpec().setAdditionalData(data);
+    extension().replace(resource);
+  }
+
+  public static void updateSpecWithNewAdditionalData(LocallyRunOperatorExtension extension,
+      String data) {
+    var resource = testResource();
+    resource.getSpec().setAdditionalData(data);
+    extension.replace(resource);
+  }
+
+  private void updateSpecWithNumber(int n) {
+    var resource = testResource();
+    resource.getSpec().setNumberOfResources(n);
+    extension().replace(resource);
+  }
+
+  public static void updateSpecWithNumber(LocallyRunOperatorExtension extension, int n) {
+    var resource = testResource();
+    resource.getSpec().setNumberOfResources(n);
+    extension.replace(resource);
+  }
+
+  abstract LocallyRunOperatorExtension extension();
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/bulkdependent/BulkExternalDependentIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/bulkdependent/BulkExternalDependentIT.java
new file mode 100644
index 0000000000..29f66e8205
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/bulkdependent/BulkExternalDependentIT.java
@@ -0,0 +1,56 @@
+package io.javaoperatorsdk.operator.bulkdependent;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension;
+import io.javaoperatorsdk.operator.sample.bulkdependent.external.ExternalBulkResourceReconciler;
+import io.javaoperatorsdk.operator.sample.bulkdependent.external.ExternalServiceMock;
+
+import static io.javaoperatorsdk.operator.bulkdependent.BulkDependentTestBase.*;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+class BulkExternalDependentIT {
+
+  @RegisterExtension
+  LocallyRunOperatorExtension extension =
+      LocallyRunOperatorExtension.builder().withReconciler(new ExternalBulkResourceReconciler())
+          .build();
+
+  ExternalServiceMock externalServiceMock = ExternalServiceMock.getInstance();
+
+  @Test
+  void managesExternalBulkResources() {
+    extension.create(testResource());
+    assertResourceNumberAndData(3, INITIAL_ADDITIONAL_DATA);
+
+    updateSpecWithNumber(extension, 1);
+    assertResourceNumberAndData(1, INITIAL_ADDITIONAL_DATA);
+
+    updateSpecWithNumber(extension, 5);
+    assertResourceNumberAndData(5, INITIAL_ADDITIONAL_DATA);
+
+    extension.delete(testResource());
+    assertResourceNumberAndData(0, INITIAL_ADDITIONAL_DATA);
+  }
+
+
+  @Test
+  void handlesResourceUpdates() {
+    extension.create(testResource());
+    assertResourceNumberAndData(3, INITIAL_ADDITIONAL_DATA);
+
+    updateSpecWithNewAdditionalData(extension, NEW_VERSION_OF_ADDITIONAL_DATA);
+    assertResourceNumberAndData(3, NEW_VERSION_OF_ADDITIONAL_DATA);
+  }
+
+  private void assertResourceNumberAndData(int n, String data) {
+    await().untilAsserted(() -> {
+      var resources = externalServiceMock.listResources();
+      assertThat(resources).hasSize(n);
+      assertThat(resources).allMatch(r -> r.getData().equals(data));
+    });
+  }
+
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/bulkdependent/ManagedBulkDependentIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/bulkdependent/ManagedBulkDependentIT.java
new file mode 100644
index 0000000000..7f074ac8f5
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/bulkdependent/ManagedBulkDependentIT.java
@@ -0,0 +1,20 @@
+package io.javaoperatorsdk.operator.bulkdependent;
+
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension;
+import io.javaoperatorsdk.operator.sample.bulkdependent.ManagedBulkDependentReconciler;
+
+class ManagedBulkDependentIT extends BulkDependentTestBase {
+
+  @RegisterExtension
+  LocallyRunOperatorExtension extension =
+      LocallyRunOperatorExtension.builder().withReconciler(new ManagedBulkDependentReconciler())
+          .build();
+
+
+  @Override
+  LocallyRunOperatorExtension extension() {
+    return extension;
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/bulkdependent/StandaloneBulkDependentIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/bulkdependent/StandaloneBulkDependentIT.java
new file mode 100644
index 0000000000..683cc1662b
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/bulkdependent/StandaloneBulkDependentIT.java
@@ -0,0 +1,19 @@
+package io.javaoperatorsdk.operator.bulkdependent;
+
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension;
+import io.javaoperatorsdk.operator.sample.bulkdependent.StandaloneBulkDependentReconciler;
+
+class StandaloneBulkDependentIT extends BulkDependentTestBase {
+
+  @RegisterExtension
+  LocallyRunOperatorExtension extension =
+      LocallyRunOperatorExtension.builder().withReconciler(new StandaloneBulkDependentReconciler())
+          .build();
+
+  @Override
+  LocallyRunOperatorExtension extension() {
+    return extension;
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/BulkDependentTestCustomResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/BulkDependentTestCustomResource.java
new file mode 100644
index 0000000000..68e6297f8c
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/BulkDependentTestCustomResource.java
@@ -0,0 +1,15 @@
+package io.javaoperatorsdk.operator.sample.bulkdependent;
+
+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("sbd")
+public class BulkDependentTestCustomResource
+    extends CustomResource<BulkDependentTestSpec, Void>
+    implements Namespaced {
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/BulkDependentTestSpec.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/BulkDependentTestSpec.java
new file mode 100644
index 0000000000..5266950b41
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/BulkDependentTestSpec.java
@@ -0,0 +1,25 @@
+package io.javaoperatorsdk.operator.sample.bulkdependent;
+
+public class BulkDependentTestSpec {
+
+  private Integer numberOfResources;
+  private String additionalData;
+
+  public Integer getNumberOfResources() {
+    return numberOfResources;
+  }
+
+  public BulkDependentTestSpec setNumberOfResources(Integer numberOfResources) {
+    this.numberOfResources = numberOfResources;
+    return this;
+  }
+
+  public BulkDependentTestSpec setAdditionalData(String additionalData) {
+    this.additionalData = additionalData;
+    return this;
+  }
+
+  public String getAdditionalData() {
+    return additionalData;
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/CRUDConfigMapBulkDependentResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/CRUDConfigMapBulkDependentResource.java
new file mode 100644
index 0000000000..83cec0bb69
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/CRUDConfigMapBulkDependentResource.java
@@ -0,0 +1,7 @@
+package io.javaoperatorsdk.operator.sample.bulkdependent;
+
+import io.javaoperatorsdk.operator.api.reconciler.dependent.GarbageCollected;
+
+public class CRUDConfigMapBulkDependentResource extends ConfigMapDeleterBulkDependentResource
+    implements GarbageCollected<BulkDependentTestCustomResource> {
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/ConfigMapDeleterBulkDependentResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/ConfigMapDeleterBulkDependentResource.java
new file mode 100644
index 0000000000..a7fbd9cb98
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/ConfigMapDeleterBulkDependentResource.java
@@ -0,0 +1,72 @@
+package io.javaoperatorsdk.operator.sample.bulkdependent;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+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.api.reconciler.ResourceDiscriminator;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.Deleter;
+import io.javaoperatorsdk.operator.processing.dependent.BulkDependentResource;
+import io.javaoperatorsdk.operator.processing.dependent.Creator;
+import io.javaoperatorsdk.operator.processing.dependent.Updater;
+import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResource;
+
+/**
+ * Not using CRUDKubernetesDependentResource so the delete functionality can be tested.
+ */
+public class ConfigMapDeleterBulkDependentResource
+    extends
+    KubernetesDependentResource<ConfigMap, BulkDependentTestCustomResource>
+    implements Creator<ConfigMap, BulkDependentTestCustomResource>,
+    Updater<ConfigMap, BulkDependentTestCustomResource>,
+    Deleter<BulkDependentTestCustomResource>,
+    BulkDependentResource<ConfigMap, BulkDependentTestCustomResource> {
+
+  public static final String LABEL_KEY = "bulk";
+  public static final String LABEL_VALUE = "true";
+  public static final String ADDITIONAL_DATA_KEY = "additionalData";
+
+  public ConfigMapDeleterBulkDependentResource() {
+    super(ConfigMap.class);
+  }
+
+  @Override
+  public ConfigMap desired(BulkDependentTestCustomResource primary,
+      int index, Context<BulkDependentTestCustomResource> context) {
+    ConfigMap configMap = new ConfigMap();
+    configMap.setMetadata(new ObjectMetaBuilder()
+        .withName(primary.getMetadata().getName() + "-" + index)
+        .withNamespace(primary.getMetadata().getNamespace())
+        .withLabels(Map.of(LABEL_KEY, LABEL_VALUE))
+        .build());
+    configMap.setData(
+        Map.of("number", "" + index, ADDITIONAL_DATA_KEY, primary.getSpec().getAdditionalData()));
+    return configMap;
+  }
+
+  @Override
+  public int count(BulkDependentTestCustomResource primary,
+      Context<BulkDependentTestCustomResource> context) {
+    return primary.getSpec().getNumberOfResources();
+  }
+
+  @Override
+  public ResourceDiscriminator<ConfigMap, BulkDependentTestCustomResource> getResourceDiscriminator(
+      int index) {
+    return (resource, primary, context) -> {
+      var resources = context.getSecondaryResources(resource).stream()
+          .filter(r -> r.getMetadata().getName().endsWith("-" + index))
+          .collect(Collectors.toList());
+      if (resources.isEmpty()) {
+        return Optional.empty();
+      } else if (resources.size() > 1) {
+        throw new IllegalStateException("More than one resource found for index:" + index);
+      } else {
+        return Optional.of(resources.get(0));
+      }
+    };
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/ManagedBulkDependentReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/ManagedBulkDependentReconciler.java
new file mode 100644
index 0000000000..3b2acd942e
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/ManagedBulkDependentReconciler.java
@@ -0,0 +1,25 @@
+package io.javaoperatorsdk.operator.sample.bulkdependent;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.Dependent;
+
+@ControllerConfiguration(dependents = @Dependent(type = CRUDConfigMapBulkDependentResource.class))
+public class ManagedBulkDependentReconciler
+    implements Reconciler<BulkDependentTestCustomResource> {
+
+  private final AtomicInteger numberOfExecutions = new AtomicInteger(0);
+
+  @Override
+  public UpdateControl<BulkDependentTestCustomResource> reconcile(
+      BulkDependentTestCustomResource resource,
+      Context<BulkDependentTestCustomResource> context) throws Exception {
+
+    numberOfExecutions.addAndGet(1);
+    return UpdateControl.noUpdate();
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/ManagedDeleterBulkReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/ManagedDeleterBulkReconciler.java
new file mode 100644
index 0000000000..e759bdd200
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/ManagedDeleterBulkReconciler.java
@@ -0,0 +1,20 @@
+package io.javaoperatorsdk.operator.sample.bulkdependent;
+
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.Dependent;
+
+@ControllerConfiguration(
+    dependents = @Dependent(type = ConfigMapDeleterBulkDependentResource.class))
+public class ManagedDeleterBulkReconciler implements Reconciler<BulkDependentTestCustomResource> {
+  @Override
+  public UpdateControl<BulkDependentTestCustomResource> reconcile(
+      BulkDependentTestCustomResource resource,
+      Context<BulkDependentTestCustomResource> context)
+      throws Exception {
+
+    return UpdateControl.noUpdate();
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/StandaloneBulkDependentReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/StandaloneBulkDependentReconciler.java
new file mode 100644
index 0000000000..4033583340
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/StandaloneBulkDependentReconciler.java
@@ -0,0 +1,58 @@
+package io.javaoperatorsdk.operator.sample.bulkdependent;
+
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.javaoperatorsdk.operator.api.reconciler.*;
+import io.javaoperatorsdk.operator.junit.KubernetesClientAware;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import io.javaoperatorsdk.operator.support.TestExecutionInfoProvider;
+
+@ControllerConfiguration
+public class StandaloneBulkDependentReconciler
+    implements Reconciler<BulkDependentTestCustomResource>, TestExecutionInfoProvider,
+    EventSourceInitializer<BulkDependentTestCustomResource>, KubernetesClientAware {
+
+  private final AtomicInteger numberOfExecutions = new AtomicInteger(0);
+
+  private ConfigMapDeleterBulkDependentResource dependent;
+  private KubernetesClient kubernetesClient;
+
+  public StandaloneBulkDependentReconciler() {
+    dependent = new CRUDConfigMapBulkDependentResource();
+  }
+
+  @Override
+  public UpdateControl<BulkDependentTestCustomResource> reconcile(
+      BulkDependentTestCustomResource resource,
+      Context<BulkDependentTestCustomResource> context) {
+    numberOfExecutions.addAndGet(1);
+
+    dependent.reconcile(resource, context);
+
+    return UpdateControl.noUpdate();
+  }
+
+  public int getNumberOfExecutions() {
+    return numberOfExecutions.get();
+  }
+
+  @Override
+  public Map<String, EventSource> prepareEventSources(
+      EventSourceContext<BulkDependentTestCustomResource> context) {
+    return EventSourceInitializer
+        .nameEventSources(dependent.initEventSource(context));
+  }
+
+  @Override
+  public KubernetesClient getKubernetesClient() {
+    return kubernetesClient;
+  }
+
+  @Override
+  public void setKubernetesClient(KubernetesClient kubernetesClient) {
+    this.kubernetesClient = kubernetesClient;
+    dependent.setKubernetesClient(kubernetesClient);
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/external/ExternalBulkDependentResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/external/ExternalBulkDependentResource.java
new file mode 100644
index 0000000000..110626a923
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/external/ExternalBulkDependentResource.java
@@ -0,0 +1,101 @@
+package io.javaoperatorsdk.operator.sample.bulkdependent.external;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ResourceDiscriminator;
+import io.javaoperatorsdk.operator.processing.dependent.*;
+import io.javaoperatorsdk.operator.processing.dependent.external.PollingDependentResource;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+import io.javaoperatorsdk.operator.sample.bulkdependent.BulkDependentTestCustomResource;
+
+public class ExternalBulkDependentResource
+    extends PollingDependentResource<ExternalResource, BulkDependentTestCustomResource>
+    implements BulkDependentResource<ExternalResource, BulkDependentTestCustomResource>,
+    BulkUpdater<ExternalResource, BulkDependentTestCustomResource> {
+
+  public static final String EXTERNAL_RESOURCE_NAME_DELIMITER = "#";
+
+  private final ExternalServiceMock externalServiceMock = ExternalServiceMock.getInstance();
+
+  public ExternalBulkDependentResource() {
+    super(ExternalResource.class, ExternalResource::getId);
+  }
+
+  @Override
+  public Map<ResourceID, Set<ExternalResource>> fetchResources() {
+    Map<ResourceID, Set<ExternalResource>> result = new HashMap<>();
+    var resources = externalServiceMock.listResources();
+    resources.forEach(er -> {
+      var resourceID = toResourceID(er);
+      result.putIfAbsent(resourceID, new HashSet<>());
+      result.get(resourceID).add(er);
+    });
+    return result;
+  }
+
+  @Override
+  public void delete(BulkDependentTestCustomResource primary,
+      Context<BulkDependentTestCustomResource> context) {
+    deleteBulkResourcesIfRequired(0, lastKnownBulkSize(), primary, context);
+  }
+
+  @Override
+  public int count(BulkDependentTestCustomResource primary,
+      Context<BulkDependentTestCustomResource> context) {
+    return primary.getSpec().getNumberOfResources();
+  }
+
+  @Override
+  public void deleteBulkResourceWithIndex(BulkDependentTestCustomResource primary,
+      ExternalResource resource, int i, Context<BulkDependentTestCustomResource> context) {
+    externalServiceMock.delete(resource.getId());
+  }
+
+  @Override
+  public ExternalResource desired(BulkDependentTestCustomResource primary, int index,
+      Context<BulkDependentTestCustomResource> context) {
+    return new ExternalResource(toExternalResourceId(primary, index),
+        primary.getSpec().getAdditionalData());
+  }
+
+  @Override
+  public ExternalResource create(ExternalResource desired, BulkDependentTestCustomResource primary,
+      Context<BulkDependentTestCustomResource> context) {
+    return externalServiceMock.create(desired);
+  }
+
+  @Override
+  public ExternalResource update(ExternalResource actual, ExternalResource desired,
+      BulkDependentTestCustomResource primary, Context<BulkDependentTestCustomResource> context) {
+    return externalServiceMock.update(desired);
+  }
+
+  @Override
+  public Matcher.Result<ExternalResource> match(ExternalResource actualResource,
+      BulkDependentTestCustomResource primary,
+      int index, Context<BulkDependentTestCustomResource> context) {
+    var desired = desired(primary, index, context);
+    return Matcher.Result.computed(desired.equals(actualResource), desired);
+  }
+
+  private static String toExternalResourceId(BulkDependentTestCustomResource primary, int i) {
+    return primary.getMetadata().getName() + EXTERNAL_RESOURCE_NAME_DELIMITER +
+        primary.getMetadata().getNamespace() +
+        EXTERNAL_RESOURCE_NAME_DELIMITER + i;
+  }
+
+  private ResourceID toResourceID(ExternalResource externalResource) {
+    var parts = externalResource.getId().split(EXTERNAL_RESOURCE_NAME_DELIMITER);
+    return new ResourceID(parts[0], parts[1]);
+  }
+
+  @Override
+  public ResourceDiscriminator<ExternalResource, BulkDependentTestCustomResource> getResourceDiscriminator(
+      int index) {
+    return (resource, primary, context) -> context.getSecondaryResources(resource).stream()
+        .filter(r -> r.getId().endsWith(EXTERNAL_RESOURCE_NAME_DELIMITER + index))
+        .collect(Collectors.toList()).stream().findFirst();
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/external/ExternalBulkResourceReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/external/ExternalBulkResourceReconciler.java
new file mode 100644
index 0000000000..2543422d74
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/external/ExternalBulkResourceReconciler.java
@@ -0,0 +1,19 @@
+package io.javaoperatorsdk.operator.sample.bulkdependent.external;
+
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.Dependent;
+import io.javaoperatorsdk.operator.sample.bulkdependent.BulkDependentTestCustomResource;
+
+@ControllerConfiguration(dependents = @Dependent(type = ExternalBulkDependentResource.class))
+public class ExternalBulkResourceReconciler implements Reconciler<BulkDependentTestCustomResource> {
+
+  @Override
+  public UpdateControl<BulkDependentTestCustomResource> reconcile(
+      BulkDependentTestCustomResource resource, Context<BulkDependentTestCustomResource> context)
+      throws Exception {
+    return UpdateControl.noUpdate();
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/external/ExternalResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/external/ExternalResource.java
new file mode 100644
index 0000000000..935fd99e47
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/external/ExternalResource.java
@@ -0,0 +1,37 @@
+package io.javaoperatorsdk.operator.sample.bulkdependent.external;
+
+import java.util.Objects;
+
+public class ExternalResource {
+
+  private String id;
+  private String data;
+
+  public ExternalResource(String id, String data) {
+    this.id = id;
+    this.data = data;
+  }
+
+  public String getId() {
+    return id;
+  }
+
+  public String getData() {
+    return data;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o)
+      return true;
+    if (o == null || getClass() != o.getClass())
+      return false;
+    ExternalResource that = (ExternalResource) o;
+    return Objects.equals(id, that.id) && Objects.equals(data, that.data);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(id, data);
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/external/ExternalServiceMock.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/external/ExternalServiceMock.java
new file mode 100644
index 0000000000..e73062ccf2
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/external/ExternalServiceMock.java
@@ -0,0 +1,39 @@
+package io.javaoperatorsdk.operator.sample.bulkdependent.external;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class ExternalServiceMock {
+
+  private static ExternalServiceMock serviceMock = new ExternalServiceMock();
+
+  private Map<String, ExternalResource> resourceMap = new ConcurrentHashMap<>();
+
+  public ExternalResource create(ExternalResource externalResource) {
+    resourceMap.put(externalResource.getId(), externalResource);
+    return externalResource;
+  }
+
+  public Optional<ExternalResource> read(String id) {
+    return Optional.ofNullable(resourceMap.get(id));
+  }
+
+  public ExternalResource update(ExternalResource externalResource) {
+    return resourceMap.put(externalResource.getId(), externalResource);
+  }
+
+  public Optional<ExternalResource> delete(String id) {
+    return Optional.ofNullable(resourceMap.remove(id));
+  }
+
+  public List<ExternalResource> listResources() {
+    return new ArrayList<>(resourceMap.values());
+  }
+
+  public static ExternalServiceMock getInstance() {
+    return serviceMock;
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/multipledependentresource/MultipleDependentResourceReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/multipledependentresource/MultipleDependentResourceReconciler.java
index 49f5ee64c1..0994e6b9b0 100644
--- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/multipledependentresource/MultipleDependentResourceReconciler.java
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/multipledependentresource/MultipleDependentResourceReconciler.java
@@ -29,7 +29,6 @@ public class MultipleDependentResourceReconciler
 
   public MultipleDependentResourceReconciler() {
     firstDependentResourceConfigMap = new MultipleDependentResourceConfigMap(FIRST_CONFIG_MAP_ID);
-
     secondDependentResourceConfigMap = new MultipleDependentResourceConfigMap(SECOND_CONFIG_MAP_ID);
 
     firstDependentResourceConfigMap