diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/AnnotationControllerConfiguration.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/AnnotationControllerConfiguration.java
index b996eeead9..4e4f945515 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/AnnotationControllerConfiguration.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/AnnotationControllerConfiguration.java
@@ -20,7 +20,6 @@
 import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.Dependent;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
-import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.AnnotationDependentResourceConfigurator;
 import io.javaoperatorsdk.operator.processing.dependent.workflow.Condition;
 import io.javaoperatorsdk.operator.processing.event.rate.RateLimiter;
 import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEventFilter;
@@ -121,19 +120,17 @@ public ResourceEventFilter<P> getEventFilter() {
     Class<ResourceEventFilter<P>>[] filterTypes =
         (Class<ResourceEventFilter<P>>[]) valueOrDefault(annotation,
             ControllerConfiguration::eventFilters, new Object[] {});
-    if (filterTypes.length > 0) {
-      for (var filterType : filterTypes) {
-        try {
-          ResourceEventFilter<P> filter = filterType.getConstructor().newInstance();
-
-          if (answer == null) {
-            answer = filter;
-          } else {
-            answer = answer.and(filter);
-          }
-        } catch (Exception e) {
-          throw new IllegalArgumentException(e);
+    for (var filterType : filterTypes) {
+      try {
+        ResourceEventFilter<P> filter = filterType.getConstructor().newInstance();
+
+        if (answer == null) {
+          answer = filter;
+        } else {
+          answer = answer.and(filter);
         }
+      } catch (Exception e) {
+        throw new IllegalArgumentException(e);
       }
     }
     return answer != null ? answer : ResourceEventFilters.passthrough();
@@ -177,22 +174,6 @@ private void configureFromAnnotatedReconciler(Object instance) {
     }
   }
 
-  @SuppressWarnings("unchecked")
-  private void configureFromCustomAnnotation(Object instance) {
-    if (instance instanceof AnnotationDependentResourceConfigurator) {
-      AnnotationDependentResourceConfigurator configurator =
-          (AnnotationDependentResourceConfigurator) instance;
-      final Class<? extends Annotation> configurationClass =
-          (Class<? extends Annotation>) Utils.getFirstTypeArgumentFromInterface(
-              instance.getClass(), AnnotationDependentResourceConfigurator.class);
-      final var configAnnotation = instance.getClass().getAnnotation(configurationClass);
-      // always called even if the annotation is null so that implementations can provide default
-      // values
-      final var config = configurator.configFrom(configAnnotation, this);
-      configurator.configureWith(config);
-    }
-  }
-
   @Override
   @SuppressWarnings("unchecked")
   public Optional<OnAddFilter<P>> onAddFilter() {
@@ -239,15 +220,10 @@ public List<DependentResourceSpec> getDependentResources() {
               "A DependentResource named '" + name + "' already exists: " + spec);
         }
 
-        final var dependentResource = Utils.instantiateAndConfigureIfNeeded(dependentType,
-            DependentResource.class,
-            Utils.contextFor(this, dependentType, Dependent.class),
-            this::configureFromCustomAnnotation);
-
         var eventSourceName = dependent.useEventSourceWithName();
         eventSourceName = Constants.NO_VALUE_SET.equals(eventSourceName) ? null : eventSourceName;
         final var context = Utils.contextFor(this, dependentType, null);
-        spec = new DependentResourceSpec(dependentResource, name,
+        spec = new DependentResourceSpec(dependentType, name,
             Set.of(dependent.dependsOn()),
             Utils.instantiate(dependent.readyPostcondition(), Condition.class, context),
             Utils.instantiate(dependent.reconcilePrecondition(), Condition.class, context),
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java
index 036aa76bbf..45d7162184 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java
@@ -145,9 +145,9 @@ default ObjectMapper getObjectMapper() {
     return Serialization.jsonMapper();
   }
 
-  @Deprecated(forRemoval = true)
+  @SuppressWarnings("rawtypes")
   default DependentResourceFactory dependentResourceFactory() {
-    return null;
+    return DependentResourceFactory.DEFAULT;
   }
 
   default Optional<LeaderElectionConfiguration> getLeaderElectionConfiguration() {
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverrider.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverrider.java
index 1f5494050e..31de28631d 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverrider.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverrider.java
@@ -1,17 +1,15 @@
 package io.javaoperatorsdk.operator.api.config;
 
 import java.time.Duration;
+import java.util.HashMap;
 import java.util.HashSet;
-import java.util.LinkedHashMap;
 import java.util.List;
-import java.util.Optional;
+import java.util.Map;
 import java.util.Set;
 import java.util.function.UnaryOperator;
-import java.util.stream.Collectors;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.config.dependent.DependentResourceSpec;
-import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DependentResourceConfigurator;
 import io.javaoperatorsdk.operator.processing.event.rate.RateLimiter;
 import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEventFilter;
 import io.javaoperatorsdk.operator.processing.event.source.filter.GenericFilter;
@@ -34,12 +32,12 @@ public class ControllerConfigurationOverrider<R extends HasMetadata> {
   private ResourceEventFilter<R> customResourcePredicate;
   private final ControllerConfiguration<R> original;
   private Duration reconciliationMaxInterval;
-  private final LinkedHashMap<String, DependentResourceSpec> namedDependentResourceSpecs;
   private OnAddFilter<R> onAddFilter;
   private OnUpdateFilter<R> onUpdateFilter;
   private GenericFilter<R> genericFilter;
   private RateLimiter rateLimiter;
   private UnaryOperator<R> cachePruneFunction;
+  private Map<DependentResourceSpec, Object> configurations;
 
   private ControllerConfigurationOverrider(ControllerConfiguration<R> original) {
     finalizer = original.getFinalizerName();
@@ -49,13 +47,9 @@ private ControllerConfigurationOverrider(ControllerConfiguration<R> original) {
     labelSelector = original.getLabelSelector();
     customResourcePredicate = original.getEventFilter();
     reconciliationMaxInterval = original.maxReconciliationInterval().orElse(null);
-    // make the original specs modifiable
-    final var dependentResources = original.getDependentResources();
-    namedDependentResourceSpecs = new LinkedHashMap<>(dependentResources.size());
     this.onAddFilter = original.onAddFilter().orElse(null);
     this.onUpdateFilter = original.onUpdateFilter().orElse(null);
     this.genericFilter = original.genericFilter().orElse(null);
-    dependentResources.forEach(drs -> namedDependentResourceSpecs.put(drs.getName(), drs));
     this.original = original;
     this.rateLimiter = original.getRateLimiter();
     this.cachePruneFunction = original.cachePruneFunction().orElse(null);
@@ -167,40 +161,24 @@ public ControllerConfigurationOverrider<R> withCachePruneFunction(
     return this;
   }
 
-  @SuppressWarnings("unchecked")
   public ControllerConfigurationOverrider<R> replacingNamedDependentResourceConfig(String name,
       Object dependentResourceConfig) {
 
-    var current = namedDependentResourceSpecs.get(name);
-    if (current == null) {
-      throw new IllegalArgumentException("Cannot find a DependentResource named: " + name);
-    }
+    final var specs = original.getDependentResources();
+    final var spec = specs.stream()
+        .filter(drs -> drs.getName().equals(name)).findFirst()
+        .orElseThrow(
+            () -> new IllegalArgumentException("Cannot find a DependentResource named: " + name));
 
-    var dependentResource = current.getDependentResource();
-    if (dependentResource instanceof DependentResourceConfigurator) {
-      var configurator = (DependentResourceConfigurator) dependentResource;
-      configurator.configureWith(dependentResourceConfig);
+    if (configurations == null) {
+      configurations = new HashMap<>(specs.size());
     }
+    configurations.put(spec, dependentResourceConfig);
 
     return this;
   }
 
   public ControllerConfiguration<R> build() {
-    final var hasModifiedNamespaces = !original.getNamespaces().equals(namespaces);
-    final var newDependentSpecs = namedDependentResourceSpecs.values().stream()
-        .peek(spec -> {
-          // if the dependent resource has a NamespaceChangeable config
-          // update the namespaces if needed, otherwise, do nothing
-          if (hasModifiedNamespaces) {
-            final Optional<?> maybeConfig = spec.getDependentResourceConfiguration();
-            maybeConfig
-                .filter(NamespaceChangeable.class::isInstance)
-                .map(NamespaceChangeable.class::cast)
-                .filter(NamespaceChangeable::allowsNamespaceChanges)
-                .ifPresent(nc -> nc.changeNamespaces(namespaces));
-          }
-        }).collect(Collectors.toList());
-
     return new DefaultControllerConfiguration<>(
         original.getAssociatedReconcilerClassName(),
         original.getName(),
@@ -217,7 +195,9 @@ public ControllerConfiguration<R> build() {
         onUpdateFilter,
         genericFilter,
         rateLimiter,
-        newDependentSpecs, cachePruneFunction);
+        original.getDependentResources(),
+        cachePruneFunction,
+        configurations);
   }
 
   public static <R extends HasMetadata> ControllerConfigurationOverrider<R> override(
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/DefaultControllerConfiguration.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/DefaultControllerConfiguration.java
index f6277c4139..6dd99ecbe2 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/DefaultControllerConfiguration.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/DefaultControllerConfiguration.java
@@ -3,11 +3,13 @@
 import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.function.UnaryOperator;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.config.dependent.DependentResourceConfigurationProvider;
 import io.javaoperatorsdk.operator.api.config.dependent.DependentResourceSpec;
 import io.javaoperatorsdk.operator.processing.event.rate.LinearRateLimiter;
 import io.javaoperatorsdk.operator.processing.event.rate.RateLimiter;
@@ -20,7 +22,7 @@
 @SuppressWarnings("rawtypes")
 public class DefaultControllerConfiguration<R extends HasMetadata>
     extends DefaultResourceConfiguration<R>
-    implements ControllerConfiguration<R> {
+    implements ControllerConfiguration<R>, DependentResourceConfigurationProvider {
 
   private final String associatedControllerClassName;
   private final String name;
@@ -32,6 +34,7 @@ public class DefaultControllerConfiguration<R extends HasMetadata>
   private final List<DependentResourceSpec> dependents;
   private final Duration reconciliationMaxInterval;
   private final RateLimiter rateLimiter;
+  private final Map<DependentResourceSpec, Object> configurations;
 
   // NOSONAR constructor is meant to provide all information
   public DefaultControllerConfiguration(
@@ -52,6 +55,31 @@ public DefaultControllerConfiguration(
       RateLimiter rateLimiter,
       List<DependentResourceSpec> dependents,
       UnaryOperator<R> cachePruneFunction) {
+    this(associatedControllerClassName, name, crdName, finalizer, generationAware, namespaces,
+        retry, labelSelector, resourceEventFilter, resourceClass, reconciliationMaxInterval,
+        onAddFilter, onUpdateFilter, genericFilter, rateLimiter, dependents, cachePruneFunction,
+        null);
+  }
+
+  DefaultControllerConfiguration(
+      String associatedControllerClassName,
+      String name,
+      String crdName,
+      String finalizer,
+      boolean generationAware,
+      Set<String> namespaces,
+      Retry retry,
+      String labelSelector,
+      ResourceEventFilter<R> resourceEventFilter,
+      Class<R> resourceClass,
+      Duration reconciliationMaxInterval,
+      OnAddFilter<R> onAddFilter,
+      OnUpdateFilter<R> onUpdateFilter,
+      GenericFilter<R> genericFilter,
+      RateLimiter rateLimiter,
+      List<DependentResourceSpec> dependents,
+      UnaryOperator<R> cachePruneFunction,
+      Map<DependentResourceSpec, Object> configurations) {
     super(labelSelector, resourceClass, onAddFilter, onUpdateFilter, genericFilter, namespaces,
         cachePruneFunction);
     this.associatedControllerClassName = associatedControllerClassName;
@@ -68,6 +96,7 @@ public DefaultControllerConfiguration(
     this.rateLimiter =
         rateLimiter != null ? rateLimiter : LinearRateLimiter.deactivatedRateLimiter();
     this.dependents = dependents != null ? dependents : Collections.emptyList();
+    this.configurations = configurations != null ? configurations : Collections.emptyMap();
   }
 
   @Override
@@ -120,4 +149,8 @@ public RateLimiter getRateLimiter() {
     return rateLimiter;
   }
 
+  @Override
+  public Object getConfigurationFor(DependentResourceSpec spec) {
+    return configurations.get(spec);
+  }
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/Utils.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/Utils.java
index fd0f612fa1..00927b7f14 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/Utils.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/Utils.java
@@ -237,7 +237,6 @@ public static String contextFor(ControllerConfiguration<?> controllerConfigurati
     }
     context += "reconciler: " + controllerConfiguration.getName();
 
-
     return context;
   }
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/dependent/ConfigurationConverter.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/dependent/ConfigurationConverter.java
new file mode 100644
index 0000000000..68e0f521de
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/dependent/ConfigurationConverter.java
@@ -0,0 +1,12 @@
+package io.javaoperatorsdk.operator.api.config.dependent;
+
+import java.lang.annotation.Annotation;
+
+import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DependentResourceConfigurator;
+
+public interface ConfigurationConverter<A extends Annotation, C, D extends DependentResourceConfigurator<? extends C>> {
+
+  C configFrom(A configAnnotation, ControllerConfiguration<?> parentConfiguration,
+      Class<D> originatingClass);
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/dependent/Configured.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/dependent/Configured.java
new file mode 100644
index 0000000000..db8c6f6db3
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/dependent/Configured.java
@@ -0,0 +1,16 @@
+package io.javaoperatorsdk.operator.api.config.dependent;
+
+import java.lang.annotation.Annotation;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+
+@Retention(RetentionPolicy.RUNTIME)
+public @interface Configured {
+
+  Class<? extends Annotation> by();
+
+  Class<?> with();
+
+  @SuppressWarnings("rawtypes")
+  Class<? extends ConfigurationConverter> converter();
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/dependent/DependentResourceConfigurationProvider.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/dependent/DependentResourceConfigurationProvider.java
new file mode 100644
index 0000000000..a0c9dc67ae
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/dependent/DependentResourceConfigurationProvider.java
@@ -0,0 +1,6 @@
+package io.javaoperatorsdk.operator.api.config.dependent;
+
+public interface DependentResourceConfigurationProvider {
+  @SuppressWarnings("rawtypes")
+  Object getConfigurationFor(DependentResourceSpec spec);
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/dependent/DependentResourceConfigurationResolver.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/dependent/DependentResourceConfigurationResolver.java
new file mode 100644
index 0000000000..9a143fc57c
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/dependent/DependentResourceConfigurationResolver.java
@@ -0,0 +1,181 @@
+package io.javaoperatorsdk.operator.api.config.dependent;
+
+import java.lang.annotation.Annotation;
+import java.util.HashMap;
+import java.util.Map;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.config.Utils;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DependentResourceConfigurator;
+
+@SuppressWarnings({"rawtypes", "unchecked"})
+public class DependentResourceConfigurationResolver {
+
+  private DependentResourceConfigurationResolver() {}
+
+  private static final Map<Class<? extends DependentResource>, ConverterAnnotationPair> converters =
+      new HashMap<>();
+  private static final Map<Class<? extends ConfigurationConverter>, ConfigurationConverter> knownConverters =
+      new HashMap<>();
+
+  public static <C extends ControllerConfiguration<? extends HasMetadata>> void configure(
+      DependentResource dependentResource, DependentResourceSpec spec, C parentConfiguration) {
+    if (dependentResource instanceof DependentResourceConfigurator) {
+      final var configurator = (DependentResourceConfigurator) dependentResource;
+      final var config = configurationFor(spec, parentConfiguration);
+      configurator.configureWith(config);
+    }
+  }
+
+  public static <C extends ControllerConfiguration<? extends HasMetadata>> Object configurationFor(
+      DependentResourceSpec spec, C parentConfiguration) {
+
+    // first check if the parent configuration has potentially already resolved the configuration
+    if (parentConfiguration instanceof DependentResourceConfigurationProvider) {
+      final var provider = (DependentResourceConfigurationProvider) parentConfiguration;
+      final var configuration = provider.getConfigurationFor(spec);
+      if (configuration != null) {
+        return configuration;
+      }
+    }
+
+    // find Configured-annotated class if it exists
+    return extractConfigurationFromConfigured(spec.getDependentResourceClass(),
+        parentConfiguration);
+  }
+
+  public static <C extends ControllerConfiguration<? extends HasMetadata>> Object extractConfigurationFromConfigured(
+      Class<? extends DependentResource> dependentResourceClass, C parentConfiguration) {
+    var converterAnnotationPair = converters.get(dependentResourceClass);
+
+    Annotation configAnnotation;
+    if (converterAnnotationPair == null) {
+      var configuredClassPair = getConfigured(dependentResourceClass);
+      if (configuredClassPair == null) {
+        return null;
+      }
+
+      // check if we already have a converter registered for the found Configured annotated class
+      converterAnnotationPair = converters.get(configuredClassPair.annotatedClass);
+      if (converterAnnotationPair == null) {
+        final var configured = configuredClassPair.configured;
+        converterAnnotationPair =
+            getOrCreateConverter(dependentResourceClass, parentConfiguration,
+                configured.converter(),
+                configured.by());
+      } else {
+        // only register the converter pair for this dependent resource class as well
+        converters.put(dependentResourceClass, converterAnnotationPair);
+      }
+    }
+
+    // find the associated configuration annotation
+    configAnnotation =
+        dependentResourceClass.getAnnotation(converterAnnotationPair.annotationClass);
+    final var converter = converterAnnotationPair.converter;
+
+    // always called even if the annotation is null so that implementations can provide default
+    // values
+    return converter.configFrom(configAnnotation, parentConfiguration, dependentResourceClass);
+  }
+
+  private static ConfiguredClassPair getConfigured(
+      Class<? extends DependentResource> dependentResourceClass) {
+    Class<? extends DependentResource> currentClass = dependentResourceClass;
+    Configured configured;
+    ConfiguredClassPair result = null;
+    while (DependentResource.class.isAssignableFrom(currentClass)) {
+      configured = currentClass.getAnnotation(Configured.class);
+      if (configured != null) {
+        result = new ConfiguredClassPair(configured, currentClass);
+        break;
+      }
+      currentClass = (Class<? extends DependentResource>) currentClass.getSuperclass();
+    }
+    return result;
+  }
+
+  private static <C extends ControllerConfiguration<? extends HasMetadata>> ConverterAnnotationPair getOrCreateConverter(
+      Class<? extends DependentResource> dependentResourceClass, C parentConfiguration,
+      Class<? extends ConfigurationConverter> converterClass,
+      Class<? extends Annotation> annotationClass) {
+    var converterPair = converters.get(dependentResourceClass);
+    if (converterPair == null) {
+      // only instantiate a new converter if we haven't done so already for this converter type
+      var converter = knownConverters.get(converterClass);
+      if (converter == null) {
+        converter = Utils.instantiate(converterClass,
+            ConfigurationConverter.class,
+            Utils.contextFor(parentConfiguration, dependentResourceClass, Configured.class));
+        knownConverters.put(converterClass, converter);
+      }
+      // record dependent class - converter association for faster future retrieval
+      converterPair = new ConverterAnnotationPair(converter, annotationClass);
+      converters.put(dependentResourceClass, converterPair);
+    }
+    return converterPair;
+  }
+
+  static ConfigurationConverter getConverter(
+      Class<? extends DependentResource> dependentResourceClass) {
+    final var converterAnnotationPair = converters.get(dependentResourceClass);
+    return converterAnnotationPair != null ? converterAnnotationPair.converter : null;
+  }
+
+  @SuppressWarnings("unused")
+  public static void registerConverter(Class<? extends DependentResource> dependentResourceClass,
+      ConfigurationConverter converter) {
+    var configured = getConfigured(dependentResourceClass);
+    if (configured == null) {
+      throw new IllegalArgumentException("There is no @" + Configured.class.getSimpleName()
+          + " annotation on " + dependentResourceClass.getName()
+          + " or its superclasses and thus doesn't need to be associated with a converter");
+    }
+
+    // find the associated configuration annotation
+    final var toRegister = new ConverterAnnotationPair(converter, configured.configured.by());
+    final Class<? extends ConfigurationConverter> converterClass = converter.getClass();
+    converters.put(dependentResourceClass, toRegister);
+
+    // also register the Configured-annotated class if not the one we're registering
+    if (!dependentResourceClass.equals(configured.annotatedClass)) {
+      converters.put(configured.annotatedClass, toRegister);
+    }
+
+    knownConverters.put(converterClass, converter);
+  }
+
+  private static class ConfiguredClassPair {
+    private final Configured configured;
+    private final Class<? extends DependentResource> annotatedClass;
+
+    private ConfiguredClassPair(Configured configured,
+        Class<? extends DependentResource> annotatedClass) {
+      this.configured = configured;
+      this.annotatedClass = annotatedClass;
+    }
+
+    @Override
+    public String toString() {
+      return annotatedClass.getName() + " -> " + configured;
+    }
+  }
+
+  private static class ConverterAnnotationPair {
+    private final ConfigurationConverter converter;
+    private final Class<? extends Annotation> annotationClass;
+
+    private ConverterAnnotationPair(ConfigurationConverter converter,
+        Class<? extends Annotation> annotationClass) {
+      this.converter = converter;
+      this.annotationClass = annotationClass;
+    }
+
+    @Override
+    public String toString() {
+      return converter.toString() + " -> " + annotationClass.getName();
+    }
+  }
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/dependent/DependentResourceSpec.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/dependent/DependentResourceSpec.java
index 5d0b8c6b01..58fd9ace4b 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/dependent/DependentResourceSpec.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/dependent/DependentResourceSpec.java
@@ -6,12 +6,11 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
-import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DependentResourceConfigurator;
 import io.javaoperatorsdk.operator.processing.dependent.workflow.Condition;
 
-public class DependentResourceSpec<R, P extends HasMetadata, C> {
+public class DependentResourceSpec<R, P extends HasMetadata> {
 
-  private final DependentResource<R, P> dependentResource;
+  private final Class<? extends DependentResource<R, P>> dependentResourceClass;
 
   private final String name;
 
@@ -25,11 +24,11 @@ public class DependentResourceSpec<R, P extends HasMetadata, C> {
 
   private final String useEventSourceWithName;
 
-  public DependentResourceSpec(DependentResource<R, P> dependentResource,
+  public DependentResourceSpec(Class<? extends DependentResource<R, P>> dependentResourceClass,
       String name, Set<String> dependsOn, Condition<?, ?> readyCondition,
       Condition<?, ?> reconcileCondition, Condition<?, ?> deletePostCondition,
       String useEventSourceWithName) {
-    this.dependentResource = dependentResource;
+    this.dependentResourceClass = dependentResourceClass;
     this.name = name;
     this.dependsOn = dependsOn;
     this.readyCondition = readyCondition;
@@ -38,28 +37,8 @@ public DependentResourceSpec(DependentResource<R, P> dependentResource,
     this.useEventSourceWithName = useEventSourceWithName;
   }
 
-  public DependentResourceSpec(DependentResourceSpec<R, P, C> other) {
-    this.dependentResource = other.dependentResource;
-    this.name = other.name;
-    this.dependsOn = other.dependsOn;
-    this.readyCondition = other.readyCondition;
-    this.reconcileCondition = other.reconcileCondition;
-    this.deletePostCondition = other.deletePostCondition;
-    this.useEventSourceWithName = other.useEventSourceWithName;
-  }
-
-  @SuppressWarnings("unchecked")
-  public Class<DependentResource<R, P>> getDependentResourceClass() {
-    return (Class<DependentResource<R, P>>) dependentResource.getClass();
-  }
-
-  @SuppressWarnings({"unchecked", "rawtypes"})
-  public Optional<C> getDependentResourceConfiguration() {
-    if (dependentResource instanceof DependentResourceConfigurator) {
-      var configurator = (DependentResourceConfigurator) dependentResource;
-      return configurator.configuration();
-    }
-    return Optional.empty();
+  public Class<? extends DependentResource<R, P>> getDependentResourceClass() {
+    return dependentResourceClass;
   }
 
   public String getName() {
@@ -80,7 +59,7 @@ public boolean equals(Object o) {
     if (o == null || getClass() != o.getClass()) {
       return false;
     }
-    DependentResourceSpec<?, ?, ?> that = (DependentResourceSpec<?, ?, ?>) o;
+    DependentResourceSpec<?, ?> that = (DependentResourceSpec<?, ?>) o;
     return name.equals(that.name);
   }
 
@@ -108,10 +87,6 @@ public Condition getDeletePostCondition() {
     return deletePostCondition;
   }
 
-  public DependentResource<R, P> getDependentResource() {
-    return dependentResource;
-  }
-
   public Optional<String> getUseEventSourceWithName() {
     return Optional.ofNullable(useEventSourceWithName);
   }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/DependentResourceFactory.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/DependentResourceFactory.java
index 139d70b002..e9e47f6d97 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/DependentResourceFactory.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/DependentResourceFactory.java
@@ -1,18 +1,21 @@
 package io.javaoperatorsdk.operator.api.reconciler.dependent;
 
+import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.config.Utils;
 import io.javaoperatorsdk.operator.api.config.dependent.DependentResourceSpec;
 
-@Deprecated
+import static io.javaoperatorsdk.operator.api.config.dependent.DependentResourceConfigurationResolver.configure;
+
 @SuppressWarnings({"rawtypes", "unchecked"})
-public interface DependentResourceFactory {
+public interface DependentResourceFactory<C extends ControllerConfiguration<?>> {
 
-  default DependentResource createFrom(DependentResourceSpec spec) {
-    return createFrom(spec.getDependentResourceClass());
-  }
+  DependentResourceFactory DEFAULT = new DependentResourceFactory() {};
 
-  default <T extends DependentResource> T createFrom(Class<T> dependentResourceClass) {
-    return (T) Utils.instantiate(dependentResourceClass, DependentResource.class, null);
+  default DependentResource createFrom(DependentResourceSpec spec, C configuration) {
+    final var dependentResourceClass = spec.getDependentResourceClass();
+    return Utils.instantiateAndConfigureIfNeeded(dependentResourceClass,
+        DependentResource.class,
+        Utils.contextFor(configuration, dependentResourceClass, Dependent.class),
+        (instance) -> configure(instance, spec, configuration));
   }
-
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/managed/AnnotationDependentResourceConfigurator.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/managed/AnnotationDependentResourceConfigurator.java
deleted file mode 100644
index d65249b753..0000000000
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/managed/AnnotationDependentResourceConfigurator.java
+++ /dev/null
@@ -1,11 +0,0 @@
-package io.javaoperatorsdk.operator.api.reconciler.dependent.managed;
-
-import java.lang.annotation.Annotation;
-
-import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
-
-public interface AnnotationDependentResourceConfigurator<A extends Annotation, C>
-    extends DependentResourceConfigurator<C> {
-
-  C configFrom(A annotation, ControllerConfiguration<?> parentConfiguration);
-}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/Controller.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/Controller.java
index 0fa167e3c2..8753e0862b 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/Controller.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/Controller.java
@@ -40,7 +40,7 @@
 import io.javaoperatorsdk.operator.api.reconciler.dependent.EventSourceReferencer;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DefaultManagedDependentResourceContext;
 import io.javaoperatorsdk.operator.health.ControllerHealthInfo;
-import io.javaoperatorsdk.operator.processing.dependent.workflow.ManagedWorkflow;
+import io.javaoperatorsdk.operator.processing.dependent.workflow.Workflow;
 import io.javaoperatorsdk.operator.processing.dependent.workflow.WorkflowCleanupResult;
 import io.javaoperatorsdk.operator.processing.event.EventProcessor;
 import io.javaoperatorsdk.operator.processing.event.EventSourceManager;
@@ -64,7 +64,7 @@ public class Controller<P extends HasMetadata>
   private final boolean contextInitializer;
   private final boolean isCleaner;
   private final Metrics metrics;
-  private final ManagedWorkflow<P> managedWorkflow;
+  private final Workflow<P> managedWorkflow;
 
   private final GroupVersionKind associatedGVK;
   private final EventProcessor<P> eventProcessor;
@@ -83,8 +83,9 @@ public Controller(Reconciler<P> reconciler,
     this.metrics = Optional.ofNullable(configurationService.getMetrics()).orElse(Metrics.NOOP);
     contextInitializer = reconciler instanceof ContextInitializer;
     isCleaner = reconciler instanceof Cleaner;
-    managedWorkflow = configurationService.getWorkflowFactory().workflowFor(configuration);
-    managedWorkflow.resolve(kubernetesClient, configuration);
+
+    final var managed = configurationService.getWorkflowFactory().workflowFor(configuration);
+    managedWorkflow = managed.resolve(kubernetesClient, configuration);
 
     eventSourceManager = new EventSourceManager<>(this);
     eventProcessor = new EventProcessor<>(eventSourceManager);
@@ -135,7 +136,7 @@ public Map<String, Object> metadata() {
           @Override
           public UpdateControl<P> execute() throws Exception {
             initContextIfNeeded(resource, context);
-            if (!managedWorkflow.isEmptyWorkflow()) {
+            if (!managedWorkflow.isEmpty()) {
               var res = managedWorkflow.reconcile(resource, context);
               ((DefaultManagedDependentResourceContext) context.managedDependentResourceContext())
                   .setWorkflowExecutionResult(res);
@@ -180,7 +181,7 @@ public Map<String, Object> metadata() {
             public DeleteControl execute() {
               initContextIfNeeded(resource, context);
               WorkflowCleanupResult workflowCleanupResult = null;
-              if (managedWorkflow.isCleaner()) {
+              if (managedWorkflow.hasCleaner()) {
                 workflowCleanupResult = managedWorkflow.cleanup(resource, context);
                 ((DefaultManagedDependentResourceContext) context.managedDependentResourceContext())
                     .setWorkflowCleanupResult(workflowCleanupResult);
@@ -231,13 +232,13 @@ public void initAndRegisterEventSources(EventSourceContext<P> context) {
     final var dependentResourcesByName = managedWorkflow.getDependentResourcesByName();
     final var size = dependentResourcesByName.size();
     if (size > 0) {
-      dependentResourcesByName.forEach((key, value) -> {
-        if (value instanceof EventSourceProvider) {
-          final var provider = (EventSourceProvider) value;
+      dependentResourcesByName.forEach((key, dependentResource) -> {
+        if (dependentResource instanceof EventSourceProvider) {
+          final var provider = (EventSourceProvider) dependentResource;
           final var source = provider.initEventSource(context);
           eventSourceManager.registerEventSource(key, source);
         } else {
-          Optional<ResourceEventSource> eventSource = value.eventSource(context);
+          Optional<ResourceEventSource> eventSource = dependentResource.eventSource(context);
           eventSource.ifPresent(es -> eventSourceManager.registerEventSource(key, es));
         }
       });
@@ -417,7 +418,7 @@ public synchronized void stop() {
   }
 
   public boolean useFinalizer() {
-    return isCleaner || managedWorkflow.isCleaner();
+    return isCleaner || managedWorkflow.hasCleaner();
   }
 
   public GroupVersionKind getAssociatedGroupVersionKind() {
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentConverter.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentConverter.java
new file mode 100644
index 0000000000..a9a60f8e0a
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentConverter.java
@@ -0,0 +1,61 @@
+package io.javaoperatorsdk.operator.processing.dependent.kubernetes;
+
+import java.util.Arrays;
+import java.util.Set;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.config.Utils;
+import io.javaoperatorsdk.operator.api.config.dependent.ConfigurationConverter;
+import io.javaoperatorsdk.operator.api.reconciler.Constants;
+import io.javaoperatorsdk.operator.api.reconciler.ResourceDiscriminator;
+import io.javaoperatorsdk.operator.processing.event.source.filter.GenericFilter;
+import io.javaoperatorsdk.operator.processing.event.source.filter.OnAddFilter;
+import io.javaoperatorsdk.operator.processing.event.source.filter.OnDeleteFilter;
+import io.javaoperatorsdk.operator.processing.event.source.filter.OnUpdateFilter;
+
+public class KubernetesDependentConverter<R extends HasMetadata, P extends HasMetadata> implements
+    ConfigurationConverter<KubernetesDependent, KubernetesDependentResourceConfig<R>, KubernetesDependentResource<R, P>> {
+
+  @Override
+  @SuppressWarnings({"unchecked", "rawtypes"})
+  public KubernetesDependentResourceConfig<R> configFrom(KubernetesDependent configAnnotation,
+      ControllerConfiguration<?> parentConfiguration,
+      Class<KubernetesDependentResource<R, P>> originatingClass) {
+    var namespaces = parentConfiguration.getNamespaces();
+    var configuredNS = false;
+    String labelSelector = null;
+    OnAddFilter<? extends HasMetadata> onAddFilter = null;
+    OnUpdateFilter<? extends HasMetadata> onUpdateFilter = null;
+    OnDeleteFilter<? extends HasMetadata> onDeleteFilter = null;
+    GenericFilter<? extends HasMetadata> genericFilter = null;
+    ResourceDiscriminator<?, ?> resourceDiscriminator = null;
+    if (configAnnotation != null) {
+      if (!Arrays.equals(KubernetesDependent.DEFAULT_NAMESPACES, configAnnotation.namespaces())) {
+        namespaces = Set.of(configAnnotation.namespaces());
+        configuredNS = true;
+      }
+
+      final var fromAnnotation = configAnnotation.labelSelector();
+      labelSelector = Constants.NO_VALUE_SET.equals(fromAnnotation) ? null : fromAnnotation;
+
+      final var context =
+          Utils.contextFor(parentConfiguration, originatingClass,
+              configAnnotation.annotationType());
+      onAddFilter = Utils.instantiate(configAnnotation.onAddFilter(), OnAddFilter.class, context);
+      onUpdateFilter =
+          Utils.instantiate(configAnnotation.onUpdateFilter(), OnUpdateFilter.class, context);
+      onDeleteFilter =
+          Utils.instantiate(configAnnotation.onDeleteFilter(), OnDeleteFilter.class, context);
+      genericFilter =
+          Utils.instantiate(configAnnotation.genericFilter(), GenericFilter.class, context);
+
+      resourceDiscriminator =
+          Utils.instantiate(configAnnotation.resourceDiscriminator(), ResourceDiscriminator.class,
+              context);
+    }
+
+    return new KubernetesDependentResourceConfig(namespaces, labelSelector, configuredNS,
+        resourceDiscriminator, onAddFilter, onUpdateFilter, onDeleteFilter, genericFilter);
+  }
+}
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 5631536da5..a8bfcb798c 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java
@@ -1,6 +1,5 @@
 package io.javaoperatorsdk.operator.processing.dependent.kubernetes;
 
-import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Optional;
 import java.util.Set;
@@ -12,35 +11,30 @@
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.fabric8.kubernetes.client.dsl.Resource;
 import io.javaoperatorsdk.operator.OperatorException;
-import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
-import io.javaoperatorsdk.operator.api.config.Utils;
+import io.javaoperatorsdk.operator.api.config.dependent.Configured;
 import io.javaoperatorsdk.operator.api.config.informer.InformerConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.Constants;
 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.ResourceDiscriminator;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.GarbageCollected;
-import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.AnnotationDependentResourceConfigurator;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DependentResourceConfigurator;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.KubernetesClientAware;
 import io.javaoperatorsdk.operator.processing.dependent.AbstractEventSourceHolderDependentResource;
 import io.javaoperatorsdk.operator.processing.dependent.Matcher;
 import io.javaoperatorsdk.operator.processing.dependent.Matcher.Result;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 import io.javaoperatorsdk.operator.processing.event.source.SecondaryToPrimaryMapper;
-import io.javaoperatorsdk.operator.processing.event.source.filter.GenericFilter;
-import io.javaoperatorsdk.operator.processing.event.source.filter.OnAddFilter;
-import io.javaoperatorsdk.operator.processing.event.source.filter.OnDeleteFilter;
-import io.javaoperatorsdk.operator.processing.event.source.filter.OnUpdateFilter;
 import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
 import io.javaoperatorsdk.operator.processing.event.source.informer.Mappers;
 
 @Ignore
-@SuppressWarnings("rawtypes")
+@Configured(by = KubernetesDependent.class, with = KubernetesDependentResourceConfig.class,
+    converter = KubernetesDependentConverter.class)
 public abstract class KubernetesDependentResource<R extends HasMetadata, P extends HasMetadata>
     extends AbstractEventSourceHolderDependentResource<R, P, InformerEventSource<R, P>>
     implements KubernetesClientAware,
-    AnnotationDependentResourceConfigurator<KubernetesDependent, KubernetesDependentResourceConfig<R>> {
+    DependentResourceConfigurator<KubernetesDependentResourceConfig<R>> {
 
   private static final Logger log = LoggerFactory.getLogger(KubernetesDependentResource.class);
 
@@ -250,46 +244,6 @@ private void cleanupAfterEventFiltering(ResourceID resourceID) {
         .cleanupOnCreateOrUpdateEventFiltering(resourceID);
   }
 
-  @Override
-  @SuppressWarnings("unchecked")
-  public KubernetesDependentResourceConfig configFrom(KubernetesDependent kubeDependent,
-      ControllerConfiguration<?> parentConfiguration) {
-    var namespaces = parentConfiguration.getNamespaces();
-    var configuredNS = false;
-    String labelSelector = null;
-    OnAddFilter<? extends HasMetadata> onAddFilter = null;
-    OnUpdateFilter<? extends HasMetadata> onUpdateFilter = null;
-    OnDeleteFilter<? extends HasMetadata> onDeleteFilter = null;
-    GenericFilter<? extends HasMetadata> genericFilter = null;
-    ResourceDiscriminator<?, ?> resourceDiscriminator = null;
-    if (kubeDependent != null) {
-      if (!Arrays.equals(KubernetesDependent.DEFAULT_NAMESPACES, kubeDependent.namespaces())) {
-        namespaces = Set.of(kubeDependent.namespaces());
-        configuredNS = true;
-      }
-
-      final var fromAnnotation = kubeDependent.labelSelector();
-      labelSelector = Constants.NO_VALUE_SET.equals(fromAnnotation) ? null : fromAnnotation;
-
-      final var context =
-          Utils.contextFor(parentConfiguration, getClass(), kubeDependent.annotationType());
-      onAddFilter = Utils.instantiate(kubeDependent.onAddFilter(), OnAddFilter.class, context);
-      onUpdateFilter =
-          Utils.instantiate(kubeDependent.onUpdateFilter(), OnUpdateFilter.class, context);
-      onDeleteFilter =
-          Utils.instantiate(kubeDependent.onDeleteFilter(), OnDeleteFilter.class, context);
-      genericFilter =
-          Utils.instantiate(kubeDependent.genericFilter(), GenericFilter.class, context);
-
-      resourceDiscriminator =
-          Utils.instantiate(kubeDependent.resourceDiscriminator(), ResourceDiscriminator.class,
-              context);
-    }
-
-    return new KubernetesDependentResourceConfig(namespaces, labelSelector, configuredNS,
-        resourceDiscriminator, onAddFilter, onUpdateFilter, onDeleteFilter, genericFilter);
-  }
-
   @Override
   public Optional<KubernetesDependentResourceConfig<R>> configuration() {
     return Optional.ofNullable(kubernetesDependentResourceConfig);
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractDependentResourceNode.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractDependentResourceNode.java
deleted file mode 100644
index 22e368249a..0000000000
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractDependentResourceNode.java
+++ /dev/null
@@ -1,103 +0,0 @@
-package io.javaoperatorsdk.operator.processing.dependent.workflow;
-
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Optional;
-
-import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
-
-@SuppressWarnings("rawtypes")
-abstract class AbstractDependentResourceNode<R, P extends HasMetadata>
-    implements DependentResourceNode<R, P> {
-
-  private final List<DependentResourceNode> dependsOn = new LinkedList<>();
-  private final List<DependentResourceNode> parents = new LinkedList<>();
-  private final String name;
-  private Condition<R, P> reconcilePrecondition;
-  private Condition<R, P> deletePostcondition;
-  private Condition<R, P> readyPostcondition;
-  private DependentResource<R, P> dependentResource;
-
-  protected AbstractDependentResourceNode(String name) {
-    this.name = name;
-  }
-
-  @Override
-  public List<? extends DependentResourceNode> getDependsOn() {
-    return dependsOn;
-  }
-
-  @Override
-  public void addParent(DependentResourceNode parent) {
-    parents.add(parent);
-  }
-
-  @Override
-  public void addDependsOnRelation(DependentResourceNode node) {
-    node.addParent(this);
-    dependsOn.add(node);
-  }
-
-  @Override
-  public List<DependentResourceNode> getParents() {
-    return parents;
-  }
-
-  @Override
-  public String getName() {
-    return name;
-  }
-
-  @Override
-  public Optional<Condition<R, P>> getReconcilePrecondition() {
-    return Optional.ofNullable(reconcilePrecondition);
-  }
-
-  @Override
-  public Optional<Condition<R, P>> getDeletePostcondition() {
-    return Optional.ofNullable(deletePostcondition);
-  }
-
-  public void setReconcilePrecondition(Condition<R, P> reconcilePrecondition) {
-    this.reconcilePrecondition = reconcilePrecondition;
-  }
-
-  public void setDeletePostcondition(Condition<R, P> cleanupCondition) {
-    this.deletePostcondition = cleanupCondition;
-  }
-
-  @Override
-  public Optional<Condition<R, P>> getReadyPostcondition() {
-    return Optional.ofNullable(readyPostcondition);
-  }
-
-  public void setReadyPostcondition(Condition<R, P> readyPostcondition) {
-    this.readyPostcondition = readyPostcondition;
-  }
-
-  public DependentResource<R, P> getDependentResource() {
-    return dependentResource;
-  }
-
-  public void setDependentResource(DependentResource<R, P> dependentResource) {
-    this.dependentResource = dependentResource;
-  }
-
-  @Override
-  public boolean equals(Object o) {
-    if (this == o) {
-      return true;
-    }
-    if (o == null || getClass() != o.getClass()) {
-      return false;
-    }
-    AbstractDependentResourceNode<?, ?> that = (AbstractDependentResourceNode<?, ?>) o;
-    return name.equals(that.name);
-  }
-
-  @Override
-  public int hashCode() {
-    return name.hashCode();
-  }
-}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java
index 354975ebad..76db792468 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java
@@ -87,7 +87,7 @@ protected boolean isInError(DependentResourceNode<?, P> dependentResourceNode) {
   protected Map<DependentResource, Exception> getErroredDependents() {
     return exceptionsDuringExecution.entrySet().stream()
         .collect(
-            Collectors.toMap(e -> workflow.getDependentResourceFor(e.getKey()), Entry::getValue));
+            Collectors.toMap(e -> e.getKey().getDependentResource(), Entry::getValue));
   }
 
   protected synchronized void handleNodeExecutionFinish(
@@ -99,11 +99,6 @@ protected synchronized void handleNodeExecutionFinish(
     }
   }
 
-  @SuppressWarnings("unchecked")
-  protected <R> DependentResource<R, P> getDependentResourceFor(DependentResourceNode<R, P> drn) {
-    return (DependentResource<R, P>) workflow.getDependentResourceFor(drn);
-  }
-
   protected <R> boolean isConditionMet(Optional<Condition<R, P>> condition,
       DependentResource<R, P> dependentResource) {
     return condition.map(c -> c.isMet(primary,
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/DefaultDependentResourceNode.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/DefaultDependentResourceNode.java
deleted file mode 100644
index d0d844ea9c..0000000000
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/DefaultDependentResourceNode.java
+++ /dev/null
@@ -1,31 +0,0 @@
-package io.javaoperatorsdk.operator.processing.dependent.workflow;
-
-import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
-
-class DefaultDependentResourceNode<R, P extends HasMetadata>
-    extends AbstractDependentResourceNode<R, P> {
-
-  public DefaultDependentResourceNode(DependentResource<R, P> dependentResource) {
-    this(dependentResource, null, null);
-  }
-
-  public DefaultDependentResourceNode(DependentResource<R, P> dependentResource,
-      Condition<R, P> reconcilePrecondition, Condition<R, P> deletePostcondition) {
-    super(getNameFor(dependentResource));
-    setDependentResource(dependentResource);
-    setReconcilePrecondition(reconcilePrecondition);
-    setDeletePostcondition(deletePostcondition);
-  }
-
-  @SuppressWarnings("rawtypes")
-  static String getNameFor(DependentResource dependentResource) {
-    return DependentResource.defaultNameFor(dependentResource.getClass()) + "#"
-        + dependentResource.hashCode();
-  }
-
-  @Override
-  public String toString() {
-    return "DependentResourceNode{" + getDependentResource() + '}';
-  }
-}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/DefaultManagedWorkflow.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/DefaultManagedWorkflow.java
index b9bc29a32f..55b9ee6edd 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/DefaultManagedWorkflow.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/DefaultManagedWorkflow.java
@@ -1,67 +1,126 @@
 package io.javaoperatorsdk.operator.processing.dependent.workflow;
 
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.KubernetesClient;
+import io.javaoperatorsdk.operator.api.config.ConfigurationServiceProvider;
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.config.dependent.DependentResourceSpec;
-import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.EventSourceReferencer;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.KubernetesClientAware;
+
+import static io.javaoperatorsdk.operator.processing.dependent.workflow.Workflow.THROW_EXCEPTION_AUTOMATICALLY_DEFAULT;
 
 @SuppressWarnings("rawtypes")
 public class DefaultManagedWorkflow<P extends HasMetadata> implements ManagedWorkflow<P> {
 
-  private final Workflow<P> workflow;
-  private final boolean isEmptyWorkflow;
-  private boolean resolved;
+  private final Set<String> topLevelResources;
+  private final Set<String> bottomLevelResources;
+  private final List<DependentResourceSpec<?, ?>> orderedSpecs;
+  private final boolean hasCleaner;
+
+  protected DefaultManagedWorkflow(List<DependentResourceSpec<?, ?>> orderedSpecs,
+      boolean hasCleaner) {
+    this.hasCleaner = hasCleaner;
+    topLevelResources = new HashSet<>(orderedSpecs.size());
+    bottomLevelResources = orderedSpecs.stream()
+        .map(DependentResourceSpec::getName)
+        .collect(Collectors.toSet());
+    this.orderedSpecs = orderedSpecs;
+    orderedSpecs.forEach(spec -> {
+      // add cycle detection?
+      if (spec.getDependsOn().isEmpty()) {
+        topLevelResources.add(spec.getName());
+      } else {
+        for (String dependsOn : spec.getDependsOn()) {
+          bottomLevelResources.remove(dependsOn);
+        }
+      }
+    });
+  }
 
-  DefaultManagedWorkflow(List<DependentResourceSpec> dependentResourceSpecs, Workflow<P> workflow) {
-    isEmptyWorkflow = dependentResourceSpecs.isEmpty();
-    this.workflow = workflow;
+  @Override
+  @SuppressWarnings("unused")
+  public List<DependentResourceSpec<?, ?>> getOrderedSpecs() {
+    return orderedSpecs;
   }
 
-  public WorkflowReconcileResult reconcile(P primary, Context<P> context) {
-    checkIfResolved();
-    return workflow.reconcile(primary, context);
+  protected Set<String> getTopLevelResources() {
+    return topLevelResources;
   }
 
-  public WorkflowCleanupResult cleanup(P primary, Context<P> context) {
-    checkIfResolved();
-    return workflow.cleanup(primary, context);
+  protected Set<String> getBottomLevelResources() {
+    return bottomLevelResources;
   }
 
-  public boolean isCleaner() {
-    return workflow.hasCleaner();
+  List<String> nodeNames() {
+    return orderedSpecs.stream().map(DependentResourceSpec::getName).collect(Collectors.toList());
   }
 
-  public boolean isEmptyWorkflow() {
-    return isEmptyWorkflow;
+  @Override
+  public boolean hasCleaner() {
+    return hasCleaner;
   }
 
-  public Map<String, DependentResource> getDependentResourcesByName() {
-    checkIfResolved();
-    final var nodes = workflow.nodes();
-    final var result = new HashMap<String, DependentResource>(nodes.size());
-    nodes.forEach((key, drn) -> result.put(key, workflow.getDependentResourceFor(drn)));
-    return result;
+  @Override
+  public boolean isEmpty() {
+    return orderedSpecs.isEmpty();
   }
 
   @Override
-  public ManagedWorkflow<P> resolve(KubernetesClient client,
+  @SuppressWarnings("unchecked")
+  public Workflow<P> resolve(KubernetesClient client,
       ControllerConfiguration<P> configuration) {
-    if (!resolved) {
-      workflow.resolve(client, configuration);
-      resolved = true;
+    final var alreadyResolved = new HashMap<String, DependentResourceNode>(orderedSpecs.size());
+    for (DependentResourceSpec spec : orderedSpecs) {
+      final var node = new DependentResourceNode(spec.getName(),
+          spec.getReconcileCondition(),
+          spec.getDeletePostCondition(),
+          spec.getReadyCondition(),
+          resolve(spec, client, configuration));
+      alreadyResolved.put(node.getName(), node);
+      spec.getDependsOn()
+          .forEach(depend -> node.addDependsOnRelation(alreadyResolved.get((String) depend)));
     }
-    return this;
+
+    final var bottom =
+        bottomLevelResources.stream().map(alreadyResolved::get).collect(Collectors.toSet());
+    final var top =
+        topLevelResources.stream().map(alreadyResolved::get).collect(Collectors.toSet());
+    return new DefaultWorkflow<>(alreadyResolved, bottom, top,
+        THROW_EXCEPTION_AUTOMATICALLY_DEFAULT, hasCleaner);
   }
 
-  private void checkIfResolved() {
-    if (!resolved) {
-      throw new IllegalStateException("resolve should be called before");
+  @SuppressWarnings({"rawtypes", "unchecked"})
+  private <R> DependentResource<R, P> resolve(DependentResourceSpec<R, P> spec,
+      KubernetesClient client,
+      ControllerConfiguration<P> configuration) {
+    final DependentResource<R, P> dependentResource =
+        ConfigurationServiceProvider.instance().dependentResourceFactory()
+            .createFrom(spec, configuration);
+
+    if (dependentResource instanceof KubernetesClientAware) {
+      ((KubernetesClientAware) dependentResource).setKubernetesClient(client);
     }
+
+    spec.getUseEventSourceWithName()
+        .ifPresent(esName -> {
+          if (dependentResource instanceof EventSourceReferencer) {
+            ((EventSourceReferencer) dependentResource).useEventSourceWithName(esName);
+          } else {
+            throw new IllegalStateException(
+                "DependentResource " + spec + " wants to use EventSource named " + esName
+                    + " but doesn't implement support for this feature by implementing "
+                    + EventSourceReferencer.class.getSimpleName());
+          }
+        });
+
+    return dependentResource;
   }
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/DefaultWorkflow.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/DefaultWorkflow.java
new file mode 100644
index 0000000000..a151ff0947
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/DefaultWorkflow.java
@@ -0,0 +1,146 @@
+package io.javaoperatorsdk.operator.processing.dependent.workflow;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.Deleter;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.GarbageCollected;
+import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResource;
+
+/**
+ * Dependents definition: so if B depends on A, the B is dependent of A.
+ *
+ * @param <P> primary resource
+ */
+@SuppressWarnings("rawtypes")
+public class DefaultWorkflow<P extends HasMetadata> implements Workflow<P> {
+
+  private final Map<String, DependentResourceNode> dependentResourceNodes;
+  private final Set<DependentResourceNode> topLevelResources;
+  private final Set<DependentResourceNode> bottomLevelResource;
+
+  private final boolean throwExceptionAutomatically;
+  private final boolean hasCleaner;
+
+  DefaultWorkflow(Set<DependentResourceNode> dependentResourceNodes) {
+    this(dependentResourceNodes, THROW_EXCEPTION_AUTOMATICALLY_DEFAULT, false);
+  }
+
+  DefaultWorkflow(Set<DependentResourceNode> dependentResourceNodes,
+      boolean throwExceptionAutomatically,
+      boolean hasCleaner) {
+    this.throwExceptionAutomatically = throwExceptionAutomatically;
+    this.hasCleaner = hasCleaner;
+
+    if (dependentResourceNodes == null) {
+      this.topLevelResources = Collections.emptySet();
+      this.bottomLevelResource = Collections.emptySet();
+      this.dependentResourceNodes = Collections.emptyMap();
+    } else {
+      this.topLevelResources = new HashSet<>(dependentResourceNodes.size());
+      this.bottomLevelResource = new HashSet<>(dependentResourceNodes);
+      this.dependentResourceNodes = toMap(dependentResourceNodes);
+    }
+  }
+
+  protected DefaultWorkflow(Map<String, DependentResourceNode> dependentResourceNodes,
+      Set<DependentResourceNode> bottomLevelResource, Set<DependentResourceNode> topLevelResources,
+      boolean throwExceptionAutomatically,
+      boolean hasCleaner) {
+    this.throwExceptionAutomatically = throwExceptionAutomatically;
+    this.hasCleaner = hasCleaner;
+
+    this.topLevelResources = topLevelResources;
+    this.bottomLevelResource = bottomLevelResource;
+    this.dependentResourceNodes = dependentResourceNodes;
+  }
+
+  @SuppressWarnings("unchecked")
+  private Map<String, DependentResourceNode> toMap(
+      Set<DependentResourceNode> dependentResourceNodes) {
+    return dependentResourceNodes.stream()
+        .peek(drn -> {
+          // add cycle detection?
+          if (drn.getDependsOn().isEmpty()) {
+            topLevelResources.add(drn);
+          } else {
+            for (DependentResourceNode dependsOn : (List<DependentResourceNode>) drn
+                .getDependsOn()) {
+              bottomLevelResource.remove(dependsOn);
+            }
+          }
+        })
+        .collect(Collectors.toMap(DependentResourceNode::getName, Function.identity()));
+  }
+
+  @Override
+  public WorkflowReconcileResult reconcile(P primary, Context<P> context) {
+    WorkflowReconcileExecutor<P> workflowReconcileExecutor =
+        new WorkflowReconcileExecutor<>(this, primary, context);
+    var result = workflowReconcileExecutor.reconcile();
+    if (throwExceptionAutomatically) {
+      result.throwAggregateExceptionIfErrorsPresent();
+    }
+    return result;
+  }
+
+  @Override
+  public WorkflowCleanupResult cleanup(P primary, Context<P> context) {
+    WorkflowCleanupExecutor<P> workflowCleanupExecutor =
+        new WorkflowCleanupExecutor<>(this, primary, context);
+    var result = workflowCleanupExecutor.cleanup();
+    if (throwExceptionAutomatically) {
+      result.throwAggregateExceptionIfErrorsPresent();
+    }
+    return result;
+  }
+
+  @Override
+  public Set<DependentResourceNode> getTopLevelDependentResources() {
+    return topLevelResources;
+  }
+
+  @Override
+  public Set<DependentResourceNode> getBottomLevelResource() {
+    return bottomLevelResource;
+  }
+
+  @Override
+  public boolean hasCleaner() {
+    return hasCleaner;
+  }
+
+  static boolean isDeletable(Class<? extends DependentResource> drClass) {
+    final var isDeleter = Deleter.class.isAssignableFrom(drClass);
+    if (!isDeleter) {
+      return false;
+    }
+
+    if (KubernetesDependentResource.class.isAssignableFrom(drClass)) {
+      return !GarbageCollected.class.isAssignableFrom(drClass);
+    }
+    return true;
+  }
+
+  @Override
+  public boolean isEmpty() {
+    return dependentResourceNodes.isEmpty();
+  }
+
+  @Override
+  public Map<String, DependentResource> getDependentResourcesByName() {
+    final var resources = new HashMap<String, DependentResource>(dependentResourceNodes.size());
+    dependentResourceNodes
+        .forEach((name, node) -> resources.put(name, node.getDependentResource()));
+    return resources;
+  }
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/DependentResourceNode.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/DependentResourceNode.java
index 11700c8d25..6d82c8cb17 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/DependentResourceNode.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/DependentResourceNode.java
@@ -1,30 +1,113 @@
 package io.javaoperatorsdk.operator.processing.dependent.workflow;
 
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Optional;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.fabric8.kubernetes.client.KubernetesClient;
-import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
 
 @SuppressWarnings("rawtypes")
-public interface DependentResourceNode<R, P extends HasMetadata> {
+public class DependentResourceNode<R, P extends HasMetadata> {
 
-  Optional<Condition<R, P>> getReconcilePrecondition();
+  private final List<DependentResourceNode> dependsOn = new LinkedList<>();
+  private final List<DependentResourceNode> parents = new LinkedList<>();
+  private final String name;
+  private Condition<R, P> reconcilePrecondition;
+  private Condition<R, P> deletePostcondition;
+  private Condition<R, P> readyPostcondition;
+  private final DependentResource<R, P> dependentResource;
 
-  Optional<Condition<R, P>> getDeletePostcondition();
+  DependentResourceNode(DependentResource<R, P> dependentResource) {
+    this(getNameFor(dependentResource), null, null, null, dependentResource);
+  }
 
-  List<? extends DependentResourceNode> getDependsOn();
+  public DependentResourceNode(String name, Condition<R, P> reconcilePrecondition,
+      Condition<R, P> deletePostcondition, Condition<R, P> readyPostcondition,
+      DependentResource<R, P> dependentResource) {
+    this.name = name;
+    this.reconcilePrecondition = reconcilePrecondition;
+    this.deletePostcondition = deletePostcondition;
+    this.readyPostcondition = readyPostcondition;
+    this.dependentResource = dependentResource;
+  }
 
-  void addDependsOnRelation(DependentResourceNode node);
+  public List<? extends DependentResourceNode> getDependsOn() {
+    return dependsOn;
+  }
 
-  Optional<Condition<R, P>> getReadyPostcondition();
+  void addParent(DependentResourceNode parent) {
+    parents.add(parent);
+  }
 
-  List<? extends DependentResourceNode> getParents();
+  void addDependsOnRelation(DependentResourceNode node) {
+    node.addParent(this);
+    dependsOn.add(node);
+  }
 
-  void addParent(DependentResourceNode parent);
+  public List<DependentResourceNode> getParents() {
+    return parents;
+  }
 
-  String getName();
+  public String getName() {
+    return name;
+  }
 
-  default void resolve(KubernetesClient client, ControllerConfiguration<P> configuration) {}
+
+  public Optional<Condition<R, P>> getReconcilePrecondition() {
+    return Optional.ofNullable(reconcilePrecondition);
+  }
+
+
+  public Optional<Condition<R, P>> getDeletePostcondition() {
+    return Optional.ofNullable(deletePostcondition);
+  }
+
+  void setReconcilePrecondition(Condition<R, P> reconcilePrecondition) {
+    this.reconcilePrecondition = reconcilePrecondition;
+  }
+
+  void setDeletePostcondition(Condition<R, P> cleanupCondition) {
+    this.deletePostcondition = cleanupCondition;
+  }
+
+  public Optional<Condition<R, P>> getReadyPostcondition() {
+    return Optional.ofNullable(readyPostcondition);
+  }
+
+  void setReadyPostcondition(Condition<R, P> readyPostcondition) {
+    this.readyPostcondition = readyPostcondition;
+  }
+
+  public DependentResource<R, P> getDependentResource() {
+    return dependentResource;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    DependentResourceNode<?, ?> that = (DependentResourceNode<?, ?>) o;
+    return name.equals(that.name);
+  }
+
+  @Override
+  public int hashCode() {
+    return name.hashCode();
+  }
+
+  @SuppressWarnings("rawtypes")
+  static String getNameFor(DependentResource dependentResource) {
+    return DependentResource.defaultNameFor(dependentResource.getClass()) + "#"
+        + dependentResource.hashCode();
+  }
+
+  @Override
+  public String toString() {
+    return "DependentResourceNode{" + getDependentResource() + '}';
+  }
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflow.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflow.java
index 518ab69289..2de3075818 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflow.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflow.java
@@ -1,58 +1,27 @@
 package io.javaoperatorsdk.operator.processing.dependent.workflow;
 
 import java.util.Collections;
-import java.util.Map;
+import java.util.List;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
-import io.javaoperatorsdk.operator.api.reconciler.Context;
-import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
+import io.javaoperatorsdk.operator.api.config.dependent.DependentResourceSpec;
 
-@SuppressWarnings("rawtypes")
 public interface ManagedWorkflow<P extends HasMetadata> {
 
-  ManagedWorkflow noOpWorkflow = new ManagedWorkflow() {
-    @Override
-    public WorkflowReconcileResult reconcile(HasMetadata primary, Context context) {
-      throw new IllegalStateException("Shouldn't be called");
-    }
+  @SuppressWarnings("unused")
+  default List<DependentResourceSpec<?, ?>> getOrderedSpecs() {
+    return Collections.emptyList();
+  }
 
-    @Override
-    public WorkflowCleanupResult cleanup(HasMetadata primary, Context context) {
-      throw new IllegalStateException("Shouldn't be called");
-    }
+  default boolean hasCleaner() {
+    return false;
+  }
 
-    @Override
-    public boolean isCleaner() {
-      return false;
-    }
+  default boolean isEmpty() {
+    return true;
+  }
 
-    @Override
-    public boolean isEmptyWorkflow() {
-      return true;
-    }
-
-    @Override
-    public Map<String, DependentResource> getDependentResourcesByName() {
-      return Collections.emptyMap();
-    }
-
-    @Override
-    public ManagedWorkflow resolve(KubernetesClient client, ControllerConfiguration configuration) {
-      return this;
-    }
-  };
-
-  WorkflowReconcileResult reconcile(P primary, Context<P> context);
-
-  WorkflowCleanupResult cleanup(P primary, Context<P> context);
-
-  boolean isCleaner();
-
-  boolean isEmptyWorkflow();
-
-  Map<String, DependentResource> getDependentResourcesByName();
-
-  ManagedWorkflow<P> resolve(KubernetesClient client, ControllerConfiguration<P> configuration);
+  Workflow<P> resolve(KubernetesClient client, ControllerConfiguration<P> configuration);
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflowFactory.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflowFactory.java
index 5b5135e95a..4923084de8 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflowFactory.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflowFactory.java
@@ -1,22 +1,35 @@
 package io.javaoperatorsdk.operator.processing.dependent.workflow;
 
+import io.fabric8.kubernetes.client.KubernetesClient;
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 
-import static io.javaoperatorsdk.operator.processing.dependent.workflow.ManagedWorkflow.noOpWorkflow;
-
-public interface ManagedWorkflowFactory {
+public interface ManagedWorkflowFactory<C extends ControllerConfiguration<?>> {
 
   @SuppressWarnings({"rawtypes", "unchecked"})
   ManagedWorkflowFactory DEFAULT = (configuration) -> {
     final var dependentResourceSpecs = configuration.getDependentResources();
     if (dependentResourceSpecs == null || dependentResourceSpecs.isEmpty()) {
-      return noOpWorkflow;
+      return new ManagedWorkflow() {
+        @Override
+        public boolean hasCleaner() {
+          return false;
+        }
+
+        @Override
+        public boolean isEmpty() {
+          return true;
+        }
+
+        @Override
+        public Workflow resolve(KubernetesClient client, ControllerConfiguration configuration) {
+          return new DefaultWorkflow(null);
+        }
+      };
     }
-    return new DefaultManagedWorkflow(dependentResourceSpecs,
-        ManagedWorkflowSupport.instance().createWorkflow(dependentResourceSpecs));
+    return ManagedWorkflowSupport.instance().createWorkflow(dependentResourceSpecs);
   };
 
   @SuppressWarnings("rawtypes")
-  ManagedWorkflow workflowFor(ControllerConfiguration configuration);
+  ManagedWorkflow workflowFor(C configuration);
 }
 
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflowSupport.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflowSupport.java
index 24525357cf..430acfd784 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflowSupport.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflowSupport.java
@@ -46,30 +46,17 @@ public void checkForNameDuplication(List<DependentResourceSpec> dependentResourc
     }
   }
 
-  public <P extends HasMetadata> Workflow<P> createWorkflow(
+
+  public <P extends HasMetadata> ManagedWorkflow<P> createWorkflow(
       List<DependentResourceSpec> dependentResourceSpecs) {
-    var orderedResourceSpecs = orderAndDetectCycles(dependentResourceSpecs);
-    final var alreadyCreated = new ArrayList<DependentResourceNode>(orderedResourceSpecs.size());
-    final boolean[] cleanerHolder = {false};
-    final var nodes = orderedResourceSpecs.stream()
-        .map(spec -> createFrom(spec, alreadyCreated, cleanerHolder))
-        .collect(Collectors.toSet());
-    return new Workflow<>(nodes, cleanerHolder[0]);
+    return createAsDefault(dependentResourceSpecs);
   }
 
-  private DependentResourceNode createFrom(DependentResourceSpec spec,
-      List<DependentResourceNode> alreadyCreated, boolean[] cleanerHolder) {
-    final var node = new SpecDependentResourceNode<>(spec);
-    alreadyCreated.add(node);
-    // if any previously checked dependent was a cleaner, no need to check further
-    cleanerHolder[0] = cleanerHolder[0] || Workflow.isDeletable(spec.getDependentResourceClass());
-    spec.getDependsOn().forEach(depend -> {
-      final DependentResourceNode dependsOn = alreadyCreated.stream()
-          .filter(drn -> depend.equals(drn.getName())).findFirst()
-          .orElseThrow();
-      node.addDependsOnRelation(dependsOn);
-    });
-    return node;
+  <P extends HasMetadata> DefaultManagedWorkflow<P> createAsDefault(
+      List<DependentResourceSpec> dependentResourceSpecs) {
+    final boolean[] cleanerHolder = {false};
+    var orderedResourceSpecs = orderAndDetectCycles(dependentResourceSpecs, cleanerHolder);
+    return new DefaultManagedWorkflow<>(orderedResourceSpecs, cleanerHolder[0]);
   }
 
   /**
@@ -77,17 +64,22 @@ private DependentResourceNode createFrom(DependentResourceSpec spec,
    * @return top-bottom ordered resources that can be added safely to workflow
    * @throws OperatorException if there is a cycle in the dependencies
    */
-  public List<DependentResourceSpec> orderAndDetectCycles(
-      List<DependentResourceSpec> dependentResourceSpecs) {
+  private List<DependentResourceSpec<?, ?>> orderAndDetectCycles(
+      List<DependentResourceSpec> dependentResourceSpecs, boolean[] cleanerHolder) {
 
     final var drInfosByName = createDRInfos(dependentResourceSpecs);
-    final var orderedSpecs = new ArrayList<DependentResourceSpec>(dependentResourceSpecs.size());
+    final var orderedSpecs =
+        new ArrayList<DependentResourceSpec<?, ?>>(dependentResourceSpecs.size());
     final var alreadyVisited = new HashSet<String>();
     var toVisit = getTopDependentResources(dependentResourceSpecs);
 
     while (!toVisit.isEmpty()) {
       final var toVisitNext = new HashSet<DependentResourceSpec>();
       toVisit.forEach(dr -> {
+        if (cleanerHolder != null) {
+          cleanerHolder[0] =
+              cleanerHolder[0] || DefaultWorkflow.isDeletable(dr.getDependentResourceClass());
+        }
         final var name = dr.getName();
         var drInfo = drInfosByName.get(name);
         if (drInfo != null) {
@@ -111,6 +103,16 @@ public List<DependentResourceSpec> orderAndDetectCycles(
     return orderedSpecs;
   }
 
+  /**
+   * @param dependentResourceSpecs list of specs
+   * @return top-bottom ordered resources that can be added safely to workflow
+   * @throws OperatorException if there is a cycle in the dependencies
+   */
+  public List<DependentResourceSpec<?, ?>> orderAndDetectCycles(
+      List<DependentResourceSpec> dependentResourceSpecs) {
+    return orderAndDetectCycles(dependentResourceSpecs, null);
+  }
+
   private static class DRInfo {
 
     private final DependentResourceSpec spec;
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/NodeExecutor.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/NodeExecutor.java
index 393e3d2e2e..0067c55321 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/NodeExecutor.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/NodeExecutor.java
@@ -17,7 +17,7 @@ protected NodeExecutor(DependentResourceNode<R, P> dependentResourceNode,
   @Override
   public void run() {
     try {
-      var dependentResource = workflowExecutor.getDependentResourceFor(dependentResourceNode);
+      var dependentResource = dependentResourceNode.getDependentResource();
 
       doRun(dependentResourceNode, dependentResource);
 
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/SpecDependentResourceNode.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/SpecDependentResourceNode.java
deleted file mode 100644
index fc6e77db6d..0000000000
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/SpecDependentResourceNode.java
+++ /dev/null
@@ -1,49 +0,0 @@
-package io.javaoperatorsdk.operator.processing.dependent.workflow;
-
-import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.fabric8.kubernetes.client.KubernetesClient;
-import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
-import io.javaoperatorsdk.operator.api.config.dependent.DependentResourceSpec;
-import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
-import io.javaoperatorsdk.operator.api.reconciler.dependent.EventSourceReferencer;
-import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.KubernetesClientAware;
-
-class SpecDependentResourceNode<R, P extends HasMetadata>
-    extends AbstractDependentResourceNode<R, P> {
-  @SuppressWarnings("unchecked")
-  public SpecDependentResourceNode(DependentResourceSpec<R, P, ?> spec) {
-    super(spec.getName());
-    setReadyPostcondition(spec.getReadyCondition());
-    setDeletePostcondition(spec.getDeletePostCondition());
-    setReconcilePrecondition(spec.getReconcileCondition());
-  }
-
-  @Override
-  @SuppressWarnings({"rawtypes", "unchecked"})
-  public void resolve(KubernetesClient client, ControllerConfiguration<P> configuration) {
-    final var spec = configuration.getDependentResources().stream()
-        .filter(drs -> drs.getName().equals(getName()))
-        .findFirst().orElseThrow();
-
-    final DependentResource<R, P> dependentResource = spec.getDependentResource();
-
-    if (dependentResource instanceof KubernetesClientAware) {
-      ((KubernetesClientAware) dependentResource).setKubernetesClient(client);
-    }
-
-    spec.getUseEventSourceWithName()
-        .ifPresent(esName -> {
-          final var name = (String) esName;
-          if (dependentResource instanceof EventSourceReferencer) {
-            ((EventSourceReferencer) dependentResource).useEventSourceWithName(name);
-          } else {
-            throw new IllegalStateException(
-                "DependentResource " + spec + " wants to use EventSource named " + name
-                    + " but doesn't implement support for this feature by implementing "
-                    + EventSourceReferencer.class.getSimpleName());
-          }
-        });
-
-    setDependentResource(dependentResource);
-  }
-}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/Workflow.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/Workflow.java
index 99e2fa94e1..c06f17b7d8 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/Workflow.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/Workflow.java
@@ -1,154 +1,45 @@
 package io.javaoperatorsdk.operator.processing.dependent.workflow;
 
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
+import java.util.Collections;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.ExecutorService;
-import java.util.function.Function;
-import java.util.stream.Collectors;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.fabric8.kubernetes.client.KubernetesClient;
-import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
-import io.javaoperatorsdk.operator.api.config.ExecutorServiceManager;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
-import io.javaoperatorsdk.operator.api.reconciler.dependent.Deleter;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
-import io.javaoperatorsdk.operator.api.reconciler.dependent.GarbageCollected;
-import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResource;
 
-/**
- * Dependents definition: so if B depends on A, the B is dependent of A.
- *
- * @param <P> primary resource
- */
-@SuppressWarnings("rawtypes")
-public class Workflow<P extends HasMetadata> {
+public interface Workflow<P extends HasMetadata> {
 
-  public static final boolean THROW_EXCEPTION_AUTOMATICALLY_DEFAULT = true;
+  boolean THROW_EXCEPTION_AUTOMATICALLY_DEFAULT = true;
 
-  private final Map<String, DependentResourceNode> dependentResourceNodes;
-  private final Set<DependentResourceNode> topLevelResources = new HashSet<>();
-  private final Set<DependentResourceNode> bottomLevelResource = new HashSet<>();
-
-  private final boolean throwExceptionAutomatically;
-  // it's "global" executor service shared between multiple reconciliations running parallel
-  private final ExecutorService executorService;
-  private boolean resolved;
-  private final boolean hasCleaner;
-
-  Workflow(Set<DependentResourceNode> dependentResourceNodes, boolean hasCleaner) {
-    this(dependentResourceNodes, ExecutorServiceManager.instance().workflowExecutorService(),
-        THROW_EXCEPTION_AUTOMATICALLY_DEFAULT, false, hasCleaner);
-  }
-
-  Workflow(Set<DependentResourceNode> dependentResourceNodes,
-      ExecutorService executorService, boolean throwExceptionAutomatically, boolean resolved,
-      boolean hasCleaner) {
-    this.executorService = executorService;
-    this.dependentResourceNodes = toMap(dependentResourceNodes);
-    this.throwExceptionAutomatically = throwExceptionAutomatically;
-    this.resolved = resolved;
-    this.hasCleaner = hasCleaner;
-  }
-
-  private Map<String, DependentResourceNode> toMap(
-      Set<DependentResourceNode> dependentResourceNodes) {
-    final var nodes = new ArrayList<>(dependentResourceNodes);
-    bottomLevelResource.addAll(nodes);
-    return dependentResourceNodes.stream()
-        .peek(drn -> {
-          // add cycle detection?
-          if (drn.getDependsOn().isEmpty()) {
-            topLevelResources.add(drn);
-          } else {
-            for (DependentResourceNode dependsOn : (List<DependentResourceNode>) drn
-                .getDependsOn()) {
-              bottomLevelResource.remove(dependsOn);
-            }
-          }
-        })
-        .collect(Collectors.toMap(DependentResourceNode::getName, Function.identity()));
+  default WorkflowReconcileResult reconcile(P primary, Context<P> context) {
+    throw new UnsupportedOperationException("Implement this");
   }
 
-  public DependentResource getDependentResourceFor(DependentResourceNode node) {
-    throwIfUnresolved();
-    return dependentResource(node);
+  default WorkflowCleanupResult cleanup(P primary, Context<P> context) {
+    throw new UnsupportedOperationException("Implement this");
   }
 
-  private DependentResource dependentResource(DependentResourceNode node) {
-    return ((AbstractDependentResourceNode) dependentResourceNodes.get(node.getName()))
-        .getDependentResource();
+  @SuppressWarnings("rawtypes")
+  default Set<DependentResourceNode> getTopLevelDependentResources() {
+    return Collections.emptySet();
   }
 
-  private void throwIfUnresolved() {
-    if (!resolved) {
-      throw new IllegalStateException(
-          "Should call resolved before trying to access DependentResources");
-    }
+  @SuppressWarnings("rawtypes")
+  default Set<DependentResourceNode> getBottomLevelResource() {
+    return Collections.emptySet();
   }
 
-  public WorkflowReconcileResult reconcile(P primary, Context<P> context) {
-    throwIfUnresolved();
-    WorkflowReconcileExecutor<P> workflowReconcileExecutor =
-        new WorkflowReconcileExecutor<>(this, primary, context);
-    var result = workflowReconcileExecutor.reconcile();
-    if (throwExceptionAutomatically) {
-      result.throwAggregateExceptionIfErrorsPresent();
-    }
-    return result;
+  default boolean hasCleaner() {
+    return false;
   }
 
-  public WorkflowCleanupResult cleanup(P primary, Context<P> context) {
-    throwIfUnresolved();
-    WorkflowCleanupExecutor<P> workflowCleanupExecutor =
-        new WorkflowCleanupExecutor<>(this, primary, context);
-    var result = workflowCleanupExecutor.cleanup();
-    if (throwExceptionAutomatically) {
-      result.throwAggregateExceptionIfErrorsPresent();
-    }
-    return result;
-  }
-
-  Set<DependentResourceNode> getTopLevelDependentResources() {
-    return topLevelResources;
-  }
-
-  Set<DependentResourceNode> getBottomLevelResource() {
-    return bottomLevelResource;
-  }
-
-  ExecutorService getExecutorService() {
-    return executorService;
-  }
-
-  Map<String, DependentResourceNode> nodes() {
-    return dependentResourceNodes;
-  }
-
-  @SuppressWarnings("unchecked")
-  void resolve(KubernetesClient client, ControllerConfiguration<P> configuration) {
-    if (!resolved) {
-      dependentResourceNodes.values().forEach(drn -> drn.resolve(client, configuration));
-      resolved = true;
-    }
-  }
-
-  boolean hasCleaner() {
-    return hasCleaner;
+  default boolean isEmpty() {
+    return true;
   }
 
-  static boolean isDeletable(Class<? extends DependentResource> drClass) {
-    final var isDeleter = Deleter.class.isAssignableFrom(drClass);
-    if (!isDeleter) {
-      return false;
-    }
-
-    if (KubernetesDependentResource.class.isAssignableFrom(drClass)) {
-      return !GarbageCollected.class.isAssignableFrom(drClass);
-    }
-    return true;
+  @SuppressWarnings("rawtypes")
+  default Map<String, DependentResource> getDependentResourcesByName() {
+    return Collections.emptyMap();
   }
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowBuilder.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowBuilder.java
index 9f24ff3bd2..c9107a4266 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowBuilder.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowBuilder.java
@@ -5,11 +5,8 @@
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.javaoperatorsdk.operator.api.config.ExecutorServiceManager;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
 
 import static io.javaoperatorsdk.operator.processing.dependent.workflow.Workflow.THROW_EXCEPTION_AUTOMATICALLY_DEFAULT;
@@ -17,14 +14,14 @@
 @SuppressWarnings({"rawtypes", "unchecked"})
 public class WorkflowBuilder<P extends HasMetadata> {
 
-  private final Map<String, DefaultDependentResourceNode<?, P>> dependentResourceNodes =
+  private final Map<String, DependentResourceNode<?, P>> dependentResourceNodes =
       new HashMap<>();
   private boolean throwExceptionAutomatically = THROW_EXCEPTION_AUTOMATICALLY_DEFAULT;
-  private DefaultDependentResourceNode currentNode;
+  private DependentResourceNode currentNode;
   private boolean isCleaner = false;
 
   public WorkflowBuilder<P> addDependentResource(DependentResource dependentResource) {
-    currentNode = new DefaultDependentResourceNode<>(dependentResource);
+    currentNode = new DependentResourceNode<>(dependentResource);
     isCleaner = dependentResource.isDeletable();
     final var name = currentNode.getName();
     dependentResourceNodes.put(name, currentNode);
@@ -64,7 +61,7 @@ public WorkflowBuilder<P> withDeletePostcondition(Condition deletePostcondition)
   DependentResourceNode getNodeByDependentResource(DependentResource<?, ?> dependentResource) {
     // first check by name
     final var node =
-        dependentResourceNodes.get(DefaultDependentResourceNode.getNameFor(dependentResource));
+        dependentResourceNodes.get(DependentResourceNode.getNameFor(dependentResource));
     if (node != null) {
       return node;
     } else {
@@ -75,26 +72,13 @@ DependentResourceNode getNodeByDependentResource(DependentResource<?, ?> depende
     }
   }
 
-  public boolean isThrowExceptionAutomatically() {
-    return throwExceptionAutomatically;
-  }
-
   public WorkflowBuilder<P> withThrowExceptionFurther(boolean throwExceptionFurther) {
     this.throwExceptionAutomatically = throwExceptionFurther;
     return this;
   }
 
   public Workflow<P> build() {
-    return build(ExecutorServiceManager.instance().workflowExecutorService());
-  }
-
-  public Workflow<P> build(int parallelism) {
-    return build(Executors.newFixedThreadPool(parallelism));
-  }
-
-  public Workflow<P> build(ExecutorService executorService) {
-    // workflow has been built from dependent resources so it is already resolved
-    return new Workflow(new HashSet<>(dependentResourceNodes.values()), executorService,
-        throwExceptionAutomatically, true, isCleaner);
+    return new DefaultWorkflow(new HashSet<>(dependentResourceNodes.values()),
+        throwExceptionAutomatically, isCleaner);
   }
 }
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 12656c2a8a..3a65d8b112 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
@@ -10,6 +10,7 @@
 import org.slf4j.LoggerFactory;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.config.ExecutorServiceManager;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.Deleter;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
@@ -52,7 +53,7 @@ private synchronized void handleCleanup(DependentResourceNode dependentResourceN
       return;
     }
 
-    Future<?> nodeFuture = workflow.getExecutorService()
+    Future<?> nodeFuture = ExecutorServiceManager.instance().workflowExecutorService()
         .submit(new CleanupExecutor<>(dependentResourceNode));
     markAsExecuting(dependentResourceNode, nodeFuture);
     log.debug("Submitted for cleanup: {}", dependentResourceNode);
@@ -116,10 +117,10 @@ private boolean hasErroredDependent(DependentResourceNode dependentResourceNode)
   private WorkflowCleanupResult createCleanupResult() {
     final var erroredDependents = getErroredDependents();
     final var postConditionNotMet = postDeleteConditionNotMet.stream()
-        .map(workflow::getDependentResourceFor)
+        .map(DependentResourceNode::getDependentResource)
         .collect(Collectors.toList());
     final var deleteCalled = this.deleteCalled.stream()
-        .map(workflow::getDependentResourceFor)
+        .map(DependentResourceNode::getDependentResource)
         .collect(Collectors.toList());
     return new WorkflowCleanupResult(erroredDependents, postConditionNotMet, deleteCalled);
   }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutor.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutor.java
index 98abe41977..773bb0332f 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutor.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutor.java
@@ -11,6 +11,7 @@
 import org.slf4j.LoggerFactory;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.config.ExecutorServiceManager;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.Deleter;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
@@ -63,11 +64,11 @@ private synchronized <R> void handleReconcile(DependentResourceNode<R, P> depend
     }
 
     boolean reconcileConditionMet = isConditionMet(dependentResourceNode.getReconcilePrecondition(),
-        getDependentResourceFor(dependentResourceNode));
+        dependentResourceNode.getDependentResource());
     if (!reconcileConditionMet) {
       handleReconcileConditionNotMet(dependentResourceNode);
     } else {
-      Future<?> nodeFuture = workflow.getExecutorService()
+      Future<?> nodeFuture = ExecutorServiceManager.instance().workflowExecutorService()
           .submit(new NodeReconcileExecutor(dependentResourceNode));
       markAsExecuting(dependentResourceNode, nodeFuture);
       log.debug("Submitted to reconcile: {}", dependentResourceNode);
@@ -85,7 +86,7 @@ private synchronized void handleDelete(DependentResourceNode dependentResourceNo
       return;
     }
 
-    Future<?> nodeFuture = workflow.getExecutorService()
+    Future<?> nodeFuture = ExecutorServiceManager.instance().workflowExecutorService()
         .submit(new NodeDeleteExecutor(dependentResourceNode));
     markAsExecuting(dependentResourceNode, nodeFuture);
     log.debug("Submitted to delete: {}", dependentResourceNode);
@@ -214,10 +215,10 @@ private boolean hasErroredParent(DependentResourceNode<?, ?> dependentResourceNo
   private WorkflowReconcileResult createReconcileResult() {
     return new WorkflowReconcileResult(
         reconciled.stream()
-            .map(workflow::getDependentResourceFor)
+            .map(DependentResourceNode::getDependentResource)
             .collect(Collectors.toList()),
         notReady.stream()
-            .map(workflow::getDependentResourceFor)
+            .map(DependentResourceNode::getDependentResource)
             .collect(Collectors.toList()),
         getErroredDependents(),
         reconcileResults);
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverriderTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverriderTest.java
index f9f75d6796..766ef6d824 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverriderTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverriderTest.java
@@ -6,8 +6,12 @@
 import org.junit.jupiter.api.Test;
 
 import io.fabric8.kubernetes.api.model.ConfigMap;
-import io.javaoperatorsdk.operator.api.reconciler.*;
+import io.javaoperatorsdk.operator.api.config.dependent.DependentResourceConfigurationResolver;
+import io.javaoperatorsdk.operator.api.reconciler.Constants;
+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.api.reconciler.dependent.DependentResource;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.ReconcileResult;
@@ -19,6 +23,7 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -107,8 +112,8 @@ private KubernetesDependentResourceConfig extractFirstDependentKubernetesResourc
 
   private Object extractDependentKubernetesResourceConfig(
       io.javaoperatorsdk.operator.api.config.ControllerConfiguration<?> configuration, int index) {
-    return configuration.getDependentResources().get(index).getDependentResourceConfiguration()
-        .orElseThrow();
+    final var spec = configuration.getDependentResources().get(index);
+    return DependentResourceConfigurationResolver.configurationFor(spec, configuration);
   }
 
   private io.javaoperatorsdk.operator.api.config.ControllerConfiguration<?> createConfiguration(
@@ -297,11 +302,12 @@ void replaceNamedDependentResourceConfigShouldWork() {
         .filter(dr -> dr.getName().equals(dependentResourceName))
         .findFirst().orElseThrow();
     assertEquals(ReadOnlyDependent.class, dependentSpec.getDependentResourceClass());
-    var maybeConfig = dependentSpec.getDependentResourceConfiguration();
-    assertTrue(maybeConfig.isPresent());
-    assertTrue(maybeConfig.get() instanceof KubernetesDependentResourceConfig);
+    var maybeConfig =
+        DependentResourceConfigurationResolver.configurationFor(dependentSpec, configuration);
+    assertNotNull(maybeConfig);
+    assertTrue(maybeConfig instanceof KubernetesDependentResourceConfig);
 
-    var config = (KubernetesDependentResourceConfig) maybeConfig.orElseThrow();
+    var config = (KubernetesDependentResourceConfig) maybeConfig;
     // check that the DependentResource inherits the controller's configuration if applicable
     assertEquals(1, config.namespaces().size());
     assertNull(config.labelSelector());
@@ -318,8 +324,8 @@ void replaceNamedDependentResourceConfigShouldWork() {
     dependents = overridden.getDependentResources();
     dependentSpec = dependents.stream().filter(dr -> dr.getName().equals(dependentResourceName))
         .findFirst().orElseThrow();
-    config = (KubernetesDependentResourceConfig) dependentSpec.getDependentResourceConfiguration()
-        .orElseThrow();
+    config = (KubernetesDependentResourceConfig) DependentResourceConfigurationResolver
+        .configurationFor(dependentSpec, overridden);
     assertEquals(1, config.namespaces().size());
     assertEquals(labelSelector, config.labelSelector());
     assertEquals(Set.of(overriddenNS), config.namespaces());
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/UtilsTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/UtilsTest.java
index 11200f55d9..ec1223377c 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/UtilsTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/UtilsTest.java
@@ -9,10 +9,8 @@
 import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
 import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
-import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.AnnotationDependentResourceConfigurator;
 import io.javaoperatorsdk.operator.processing.dependent.EmptyTestDependentResource;
 import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResource;
-import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResourceConfig;
 import io.javaoperatorsdk.operator.sample.simple.TestCustomResource;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -96,10 +94,6 @@ void getsFirstTypeArgumentFromInterface() {
     assertThatIllegalArgumentException().isThrownBy(
         () -> Utils.getFirstTypeArgumentFromInterface(TestKubernetesDependentResource.class,
             DependentResource.class));
-
-    assertThat(Utils.getTypeArgumentFromInterfaceByIndex(TestKubernetesDependentResource.class,
-        AnnotationDependentResourceConfigurator.class, 1))
-        .isEqualTo(KubernetesDependentResourceConfig.class);
   }
 
   @Test
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/dependent/DependentResourceConfigurationResolverTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/dependent/DependentResourceConfigurationResolverTest.java
new file mode 100644
index 0000000000..983215bcb9
--- /dev/null
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/dependent/DependentResourceConfigurationResolverTest.java
@@ -0,0 +1,209 @@
+package io.javaoperatorsdk.operator.api.config.dependent;
+
+import java.lang.annotation.Annotation;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.util.Optional;
+
+import org.junit.jupiter.api.Test;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.api.model.Service;
+import io.javaoperatorsdk.operator.api.config.AnnotationControllerConfiguration;
+import io.javaoperatorsdk.operator.api.config.ControllerConfigurationOverrider;
+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.api.reconciler.dependent.DependentResource;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.ReconcileResult;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DependentResourceConfigurator;
+import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentConverter;
+import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResource;
+
+import static io.javaoperatorsdk.operator.api.config.dependent.DependentResourceConfigurationResolverTest.CustomAnnotationReconciler.DR_NAME;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class DependentResourceConfigurationResolverTest {
+
+  @Test
+  void controllerConfigurationProvidedShouldBeReturnedIfAvailable() {
+    final var cfg = new AnnotationControllerConfiguration<>(new CustomAnnotationReconciler());
+    final var customConfig = DependentResourceConfigurationResolver
+        .extractConfigurationFromConfigured(CustomAnnotatedDep.class, cfg);
+    assertTrue(customConfig instanceof CustomConfig);
+    assertEquals(CustomAnnotatedDep.PROVIDED_VALUE, ((CustomConfig) customConfig).getValue());
+    final var newConfig = new CustomConfig(72);
+    final var overridden = ControllerConfigurationOverrider.override(cfg)
+        .replacingNamedDependentResourceConfig(DR_NAME, newConfig)
+        .build();
+    final var spec = cfg.getDependentResources().stream()
+        .filter(s -> DR_NAME.equals(s.getName()))
+        .findFirst()
+        .orElseThrow();
+    assertEquals(newConfig,
+        DependentResourceConfigurationResolver.configurationFor(spec, overridden));
+  }
+
+  @Test
+  void getConverterShouldWork() {
+    final var cfg = new AnnotationControllerConfiguration<>(new CustomAnnotationReconciler());
+    var converter = DependentResourceConfigurationResolver.getConverter(CustomAnnotatedDep.class);
+    assertNull(converter);
+    assertNull(DependentResourceConfigurationResolver.getConverter(ChildCustomAnnotatedDep.class));
+
+    // extracting configuration should trigger converter creation
+    DependentResourceConfigurationResolver.extractConfigurationFromConfigured(
+        CustomAnnotatedDep.class, cfg);
+    converter = DependentResourceConfigurationResolver.getConverter(CustomAnnotatedDep.class);
+    assertNotNull(converter);
+    assertEquals(CustomConfigConverter.class, converter.getClass());
+
+    converter = DependentResourceConfigurationResolver.getConverter(ChildCustomAnnotatedDep.class);
+    assertNull(converter);
+    DependentResourceConfigurationResolver.extractConfigurationFromConfigured(
+        ChildCustomAnnotatedDep.class, cfg);
+    converter = DependentResourceConfigurationResolver.getConverter(ChildCustomAnnotatedDep.class);
+    assertNotNull(converter);
+    assertEquals(CustomConfigConverter.class, converter.getClass());
+    assertEquals(DependentResourceConfigurationResolver.getConverter(CustomAnnotatedDep.class),
+        converter);
+  }
+
+  @SuppressWarnings("rawtypes")
+  @Test
+  void registerConverterShouldWork() {
+    final var cfg = new AnnotationControllerConfiguration<>(new CustomAnnotationReconciler());
+    var converter = DependentResourceConfigurationResolver.getConverter(ConfigMapDep.class);
+    assertNull(converter);
+    DependentResourceConfigurationResolver.extractConfigurationFromConfigured(ConfigMapDep.class,
+        cfg);
+    converter = DependentResourceConfigurationResolver.getConverter(ConfigMapDep.class);
+    assertTrue(converter instanceof KubernetesDependentConverter);
+    final var overriddenConverter = new ConfigurationConverter() {
+      @Override
+      public Object configFrom(Annotation configAnnotation,
+          io.javaoperatorsdk.operator.api.config.ControllerConfiguration parentConfiguration,
+          Class originatingClass) {
+        return null;
+      }
+    };
+    DependentResourceConfigurationResolver.registerConverter(KubernetesDependentResource.class,
+        overriddenConverter);
+
+    // already resolved converters are kept unchanged
+    converter = DependentResourceConfigurationResolver.getConverter(ConfigMapDep.class);
+    assertTrue(converter instanceof KubernetesDependentConverter);
+
+    // but new converters should use the overridden version
+    DependentResourceConfigurationResolver.extractConfigurationFromConfigured(ServiceDep.class,
+        cfg);
+    converter = DependentResourceConfigurationResolver.getConverter(ServiceDep.class);
+    assertEquals(overriddenConverter, converter);
+  }
+
+  @ControllerConfiguration(dependents = {
+      @Dependent(type = CustomAnnotatedDep.class, name = DR_NAME),
+      @Dependent(type = ChildCustomAnnotatedDep.class),
+      @Dependent(type = ConfigMapDep.class),
+      @Dependent(type = ServiceDep.class)
+  })
+  static class CustomAnnotationReconciler implements Reconciler<ConfigMap> {
+
+    public static final String DR_NAME = "first";
+
+    @Override
+    public UpdateControl<ConfigMap> reconcile(ConfigMap resource, Context<ConfigMap> context)
+        throws Exception {
+      return null;
+    }
+  }
+
+  private static class ConfigMapDep extends KubernetesDependentResource<ConfigMap, ConfigMap> {
+
+    public ConfigMapDep() {
+      super(ConfigMap.class);
+    }
+  }
+
+  private static class ServiceDep extends KubernetesDependentResource<Service, ConfigMap> {
+
+    public ServiceDep() {
+      super(Service.class);
+    }
+  }
+
+  @CustomAnnotation(value = CustomAnnotatedDep.PROVIDED_VALUE)
+  @Configured(by = CustomAnnotation.class, with = CustomConfig.class,
+      converter = CustomConfigConverter.class)
+  private static class CustomAnnotatedDep implements DependentResource<ConfigMap, ConfigMap>,
+      DependentResourceConfigurator<CustomConfig> {
+
+    public static final int PROVIDED_VALUE = 42;
+    private CustomConfig config;
+
+    @Override
+    public ReconcileResult<ConfigMap> reconcile(ConfigMap primary, Context<ConfigMap> context) {
+      return null;
+    }
+
+    @Override
+    public Class<ConfigMap> resourceType() {
+      return ConfigMap.class;
+    }
+
+    @Override
+    public void configureWith(CustomConfig config) {
+      this.config = config;
+    }
+
+    @Override
+    public Optional<CustomConfig> configuration() {
+      return Optional.ofNullable(config);
+    }
+  }
+
+  private static class ChildCustomAnnotatedDep extends CustomAnnotatedDep {
+
+  }
+
+  @Retention(RetentionPolicy.RUNTIME)
+  private @interface CustomAnnotation {
+
+    int value();
+  }
+
+  private static class CustomConfig {
+
+    private final int value;
+
+    private CustomConfig(int value) {
+      this.value = value;
+    }
+
+    public int getValue() {
+      return value;
+    }
+  }
+
+  private static class CustomConfigConverter
+      implements ConfigurationConverter<CustomAnnotation, CustomConfig, CustomAnnotatedDep> {
+
+    static final int CONVERTER_PROVIDED_DEFAULT = 7;
+
+    @Override
+    public CustomConfig configFrom(CustomAnnotation configAnnotation,
+        io.javaoperatorsdk.operator.api.config.ControllerConfiguration<?> parentConfiguration,
+        Class<CustomAnnotatedDep> originatingClass) {
+      if (configAnnotation == null) {
+        return new CustomConfig(CONVERTER_PROVIDED_DEFAULT);
+      } else {
+        return new CustomConfig(configAnnotation.value());
+      }
+    }
+  }
+}
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflowSupportTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflowSupportTest.java
index 3d5bde29db..0f6fea2449 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflowSupportTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflowSupportTest.java
@@ -138,16 +138,12 @@ void createsWorkflow() {
         createDRS(NAME_3, NAME_1),
         createDRS(NAME_4, NAME_3, NAME_2));
 
-    var workflow = managedWorkflowSupport.createWorkflow(specs);
+    var workflow = managedWorkflowSupport.createAsDefault(specs);
 
-    assertThat(workflow.nodes().values()).map(DependentResourceNode::getName)
+    assertThat(workflow.nodeNames())
         .containsExactlyInAnyOrder(NAME_1, NAME_2, NAME_3, NAME_4);
-    assertThat(workflow.getTopLevelDependentResources())
-        .map(DependentResourceNode::getName)
-        .containsExactly(NAME_1);
-    assertThat(workflow.getBottomLevelResource())
-        .map(DependentResourceNode::getName)
-        .containsExactly(NAME_4);
+    assertThat(workflow.getTopLevelResources()).containsExactly(NAME_1);
+    assertThat(workflow.getBottomLevelResources()).containsExactly(NAME_4);
   }
 
 }
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflowTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflowTest.java
index d8771af5e5..68e3068de2 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflowTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflowTest.java
@@ -23,19 +23,19 @@ class ManagedWorkflowTest {
 
   @Test
   void checksIfWorkflowEmpty() {
-    assertThat(managedWorkflow().isEmptyWorkflow()).isTrue();
-    assertThat(managedWorkflow(createDRS(NAME)).isEmptyWorkflow()).isFalse();
+    assertThat(managedWorkflow().isEmpty()).isTrue();
+    assertThat(managedWorkflow(createDRS(NAME)).isEmpty()).isFalse();
   }
 
   @Test
   void isNotCleanerIfNoDeleter() {
-    assertThat(managedWorkflow(createDRS(NAME)).isCleaner()).isFalse();
+    assertThat(managedWorkflow(createDRS(NAME)).hasCleaner()).isFalse();
   }
 
   @Test
   void isNotCleanerIfGarbageCollected() {
     assertThat(managedWorkflow(createDRSWithTraits(NAME, GarbageCollected.class))
-        .isCleaner()).isFalse();
+        .hasCleaner()).isFalse();
   }
 
   @Test
@@ -43,18 +43,18 @@ void isCleanerShouldWork() {
     assertThat(managedWorkflow(
         createDRSWithTraits(NAME, GarbageCollected.class),
         createDRSWithTraits("foo", Deleter.class))
-        .isCleaner()).isTrue();
+        .hasCleaner()).isTrue();
 
     assertThat(managedWorkflow(
         createDRSWithTraits("foo", Deleter.class),
         createDRSWithTraits(NAME, GarbageCollected.class))
-        .isCleaner()).isTrue();
+        .hasCleaner()).isTrue();
   }
 
   @Test
   void isCleanerIfHasDeleter() {
     var spec = createDRSWithTraits(NAME, Deleter.class);
-    assertThat(managedWorkflow(spec).isCleaner()).isTrue();
+    assertThat(managedWorkflow(spec).hasCleaner()).isTrue();
   }
 
   ManagedWorkflow managedWorkflow(DependentResourceSpec... specs) {
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflowTestUtils.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflowTestUtils.java
index 9b97963366..25c0ad139b 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflowTestUtils.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/ManagedWorkflowTestUtils.java
@@ -20,7 +20,7 @@ public class ManagedWorkflowTestUtils {
 
   @SuppressWarnings("unchecked")
   public static DependentResourceSpec createDRS(String name, String... dependOns) {
-    return new DependentResourceSpec(new EmptyTestDependentResource(), name, Set.of(dependOns),
+    return new DependentResourceSpec(EmptyTestDependentResource.class, name, Set.of(dependOns),
         null, null, null, null);
   }
 
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowTest.java
index 382022782e..4e6cc5c329 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowTest.java
@@ -34,7 +34,7 @@ void calculatesTopLevelResources() {
 
     Set<DependentResource> topResources =
         workflow.getTopLevelDependentResources().stream()
-            .map(workflow::getDependentResourceFor)
+            .map(DependentResourceNode::getDependentResource)
             .collect(Collectors.toSet());
 
     assertThat(topResources).containsExactlyInAnyOrder(dr1, independentDR);
@@ -54,7 +54,7 @@ void calculatesBottomLevelResources() {
 
     Set<DependentResource> bottomResources =
         workflow.getBottomLevelResource().stream()
-            .map(workflow::getDependentResourceFor)
+            .map(DependentResourceNode::getDependentResource)
             .collect(Collectors.toSet());
 
     assertThat(bottomResources).containsExactlyInAnyOrder(dr2, independentDR);
@@ -64,19 +64,19 @@ void calculatesBottomLevelResources() {
   @Test
   void isDeletableShouldWork() {
     var dr = mock(DependentResource.class);
-    assertFalse(Workflow.isDeletable(dr.getClass()));
+    assertFalse(DefaultWorkflow.isDeletable(dr.getClass()));
 
     dr = mock(DependentResource.class, withSettings().extraInterfaces(Deleter.class));
-    assertTrue(Workflow.isDeletable(dr.getClass()));
+    assertTrue(DefaultWorkflow.isDeletable(dr.getClass()));
 
     dr = mock(KubernetesDependentResource.class);
-    assertFalse(Workflow.isDeletable(dr.getClass()));
+    assertFalse(DefaultWorkflow.isDeletable(dr.getClass()));
 
     dr = mock(KubernetesDependentResource.class, withSettings().extraInterfaces(Deleter.class));
-    assertTrue(Workflow.isDeletable(dr.getClass()));
+    assertTrue(DefaultWorkflow.isDeletable(dr.getClass()));
 
     dr = mock(KubernetesDependentResource.class, withSettings().extraInterfaces(Deleter.class,
         GarbageCollected.class));
-    assertFalse(Workflow.isDeletable(dr.getClass()));
+    assertFalse(DefaultWorkflow.isDeletable(dr.getClass()));
   }
 }
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/config/runtime/AnnotationControllerConfigurationTest.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/config/runtime/AnnotationControllerConfigurationTest.java
index daf1964d13..bbc49ebf33 100644
--- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/config/runtime/AnnotationControllerConfigurationTest.java
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/config/runtime/AnnotationControllerConfigurationTest.java
@@ -16,6 +16,9 @@
 import io.fabric8.kubernetes.api.model.ConfigMap;
 import io.javaoperatorsdk.operator.OperatorException;
 import io.javaoperatorsdk.operator.api.config.AnnotationConfigurable;
+import io.javaoperatorsdk.operator.api.config.dependent.ConfigurationConverter;
+import io.javaoperatorsdk.operator.api.config.dependent.Configured;
+import io.javaoperatorsdk.operator.api.config.dependent.DependentResourceConfigurationResolver;
 import io.javaoperatorsdk.operator.api.config.dependent.DependentResourceSpec;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
@@ -24,6 +27,8 @@
 import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.Dependent;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.ReconcileResult;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DependentResourceConfigurator;
 import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependent;
 import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResource;
 import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResourceConfig;
@@ -58,9 +63,9 @@ void defaultValuesShouldBeConsistent() {
   @SuppressWarnings("rawtypes")
   private KubernetesDependentResourceConfig extractDependentKubernetesResourceConfig(
       io.javaoperatorsdk.operator.api.config.ControllerConfiguration<?> configuration, int index) {
-    return (KubernetesDependentResourceConfig) configuration.getDependentResources().get(index)
-        .getDependentResourceConfiguration()
-        .orElseThrow();
+    final var spec = configuration.getDependentResources().get(index);
+    return (KubernetesDependentResourceConfig) DependentResourceConfigurationResolver
+        .configurationFor(spec, configuration);
   }
 
   @Test
@@ -78,10 +83,11 @@ void getDependentResources() {
     assertTrue(dependents.stream().anyMatch(d -> d.getName().equals(dependentResourceName)));
     var dependentSpec = findByName(dependents, dependentResourceName);
     assertEquals(ReadOnlyDependent.class, dependentSpec.getDependentResourceClass());
-    var maybeConfig = dependentSpec.getDependentResourceConfiguration();
-    assertTrue(maybeConfig.isPresent());
-    assertTrue(maybeConfig.get() instanceof KubernetesDependentResourceConfig);
-    final var config = (KubernetesDependentResourceConfig) maybeConfig.orElseThrow();
+    var maybeConfig =
+        DependentResourceConfigurationResolver.configurationFor(dependentSpec, configuration);
+    assertNotNull(maybeConfig);
+    assertTrue(maybeConfig instanceof KubernetesDependentResourceConfig);
+    final var config = (KubernetesDependentResourceConfig) maybeConfig;
     // check that the DependentResource inherits the controller's configuration if applicable
     assertEquals(1, config.namespaces().size());
     assertEquals(Set.of(OneDepReconciler.CONFIGURED_NS), config.namespaces());
@@ -92,9 +98,10 @@ void getDependentResources() {
     assertEquals(1, dependents.size());
     dependentSpec = findByName(dependents, NamedDepReconciler.NAME);
     assertEquals(ReadOnlyDependent.class, dependentSpec.getDependentResourceClass());
-    maybeConfig = dependentSpec.getDependentResourceConfiguration();
-    assertTrue(maybeConfig.isPresent());
-    assertTrue(maybeConfig.get() instanceof KubernetesDependentResourceConfig);
+    maybeConfig = DependentResourceConfigurationResolver.configurationFor(dependentSpec,
+        configuration);
+    assertNotNull(maybeConfig);
+    assertTrue(maybeConfig instanceof KubernetesDependentResourceConfig);
   }
 
   @Test
@@ -183,6 +190,20 @@ void controllerConfigurationOnSuperClassShouldWork() {
     assertNotNull(config.getName());
   }
 
+  @Test
+  void configuringFromCustomAnnotationsShouldWork() {
+    var config = new AnnotationControllerConfiguration<>(new CustomAnnotationReconciler());
+    assertEquals(CustomAnnotatedDep.PROVIDED_VALUE, getValue(config, 0));
+    assertEquals(CustomConfigConverter.CONVERTER_PROVIDED_DEFAULT, getValue(config, 1));
+  }
+
+  private static int getValue(
+      io.javaoperatorsdk.operator.api.config.ControllerConfiguration<?> configuration, int index) {
+    return ((CustomConfig) DependentResourceConfigurationResolver
+        .configurationFor(configuration.getDependentResources().get(index), configuration))
+        .getValue();
+  }
+
   @ControllerConfiguration(
       maxReconciliationInterval = @MaxReconciliationInterval(interval = 50,
           timeUnit = TimeUnit.SECONDS))
@@ -353,4 +374,84 @@ public UpdateControl<ConfigMap> reconcile(ConfigMap resource, Context<ConfigMap>
       return null;
     }
   }
+
+  @ControllerConfiguration(dependents = {
+      @Dependent(type = CustomAnnotatedDep.class),
+      @Dependent(type = ChildCustomAnnotatedDep.class)
+  })
+  private static class CustomAnnotationReconciler implements Reconciler<ConfigMap> {
+
+    @Override
+    public UpdateControl<ConfigMap> reconcile(ConfigMap resource, Context<ConfigMap> context)
+        throws Exception {
+      return null;
+    }
+  }
+
+  @CustomAnnotation(value = CustomAnnotatedDep.PROVIDED_VALUE)
+  @Configured(by = CustomAnnotation.class, with = CustomConfig.class,
+      converter = CustomConfigConverter.class)
+  private static class CustomAnnotatedDep implements DependentResource<ConfigMap, ConfigMap>,
+      DependentResourceConfigurator<CustomConfig> {
+
+    public static final int PROVIDED_VALUE = 42;
+    private CustomConfig config;
+
+    @Override
+    public ReconcileResult<ConfigMap> reconcile(ConfigMap primary, Context<ConfigMap> context) {
+      return null;
+    }
+
+    @Override
+    public Class<ConfigMap> resourceType() {
+      return ConfigMap.class;
+    }
+
+    @Override
+    public void configureWith(CustomConfig config) {
+      this.config = config;
+    }
+
+    @Override
+    public Optional<CustomConfig> configuration() {
+      return Optional.ofNullable(config);
+    }
+  }
+
+  private static class ChildCustomAnnotatedDep extends CustomAnnotatedDep {
+
+  }
+
+  @Retention(RetentionPolicy.RUNTIME)
+  private @interface CustomAnnotation {
+    int value();
+  }
+
+  private static class CustomConfig {
+    private final int value;
+
+    private CustomConfig(int value) {
+      this.value = value;
+    }
+
+    public int getValue() {
+      return value;
+    }
+  }
+
+  private static class CustomConfigConverter
+      implements ConfigurationConverter<CustomAnnotation, CustomConfig, CustomAnnotatedDep> {
+    static final int CONVERTER_PROVIDED_DEFAULT = 7;
+
+    @Override
+    public CustomConfig configFrom(CustomAnnotation configAnnotation,
+        io.javaoperatorsdk.operator.api.config.ControllerConfiguration<?> parentConfiguration,
+        Class<CustomAnnotatedDep> originatingClass) {
+      if (configAnnotation == null) {
+        return new CustomConfig(CONVERTER_PROVIDED_DEFAULT);
+      } else {
+        return new CustomConfig(configAnnotation.value());
+      }
+    }
+  }
 }
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
index a572572a97..ebc1655c38 100644
--- 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
@@ -5,21 +5,24 @@
 
 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.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceInitializer;
+import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
 import io.javaoperatorsdk.operator.api.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))
+    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
diff --git a/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/dependent/SchemaDependentResource.java b/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/dependent/SchemaDependentResource.java
index 8fb8855e41..d7ee5d368f 100644
--- a/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/dependent/SchemaDependentResource.java
+++ b/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/dependent/SchemaDependentResource.java
@@ -13,13 +13,16 @@
 
 import io.fabric8.kubernetes.api.model.Secret;
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.config.dependent.ConfigurationConverter;
+import io.javaoperatorsdk.operator.api.config.dependent.Configured;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.Deleter;
-import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.AnnotationDependentResourceConfigurator;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DependentResourceConfigurator;
 import io.javaoperatorsdk.operator.processing.dependent.Creator;
 import io.javaoperatorsdk.operator.processing.dependent.external.PerResourcePollingDependentResource;
 import io.javaoperatorsdk.operator.sample.MySQLDbConfig;
 import io.javaoperatorsdk.operator.sample.MySQLSchema;
+import io.javaoperatorsdk.operator.sample.dependent.SchemaDependentResource.ResourcePollerConfigConverter;
 import io.javaoperatorsdk.operator.sample.schema.Schema;
 import io.javaoperatorsdk.operator.sample.schema.SchemaService;
 
@@ -30,9 +33,11 @@
 @SchemaConfig(pollPeriod = 700, host = "127.0.0.1",
     port = SchemaDependentResource.LOCAL_PORT,
     user = "root", password = "password") // NOSONAR: password is only used locally, example only
+@Configured(by = SchemaConfig.class, with = ResourcePollerConfig.class,
+    converter = ResourcePollerConfigConverter.class)
 public class SchemaDependentResource
     extends PerResourcePollingDependentResource<Schema, MySQLSchema>
-    implements AnnotationDependentResourceConfigurator<SchemaConfig, ResourcePollerConfig>,
+    implements DependentResourceConfigurator<ResourcePollerConfig>,
     Creator<Schema, MySQLSchema>, Deleter<MySQLSchema> {
 
   public static final String NAME = "schema";
@@ -56,18 +61,6 @@ public void configureWith(ResourcePollerConfig config) {
     setPollingPeriod(config.getPollPeriod());
   }
 
-  @Override
-  public ResourcePollerConfig configFrom(SchemaConfig annotation,
-      ControllerConfiguration<?> parentConfiguration) {
-    if (annotation != null) {
-      return new ResourcePollerConfig(annotation.pollPeriod(),
-          new MySQLDbConfig(annotation.host(), "" + annotation.port(),
-              annotation.user(), annotation.password()));
-    }
-    return new ResourcePollerConfig(SchemaConfig.DEFAULT_POLL_PERIOD,
-        MySQLDbConfig.loadFromEnvironmentVars());
-  }
-
   @Override
   public Schema desired(MySQLSchema primary, Context<MySQLSchema> context) {
     return new Schema(primary.getMetadata().getName(), primary.getSpec().getEncoding());
@@ -120,4 +113,21 @@ public Set<Schema> fetchResources(MySQLSchema primaryResource) {
       throw new RuntimeException("Error while trying read Schema", e);
     }
   }
+
+  static class ResourcePollerConfigConverter implements
+      ConfigurationConverter<SchemaConfig, ResourcePollerConfig, SchemaDependentResource> {
+
+    @Override
+    public ResourcePollerConfig configFrom(SchemaConfig configAnnotation,
+        ControllerConfiguration<?> parentConfiguration,
+        Class<SchemaDependentResource> originatingClass) {
+      if (configAnnotation != null) {
+        return new ResourcePollerConfig(configAnnotation.pollPeriod(),
+            new MySQLDbConfig(configAnnotation.host(), "" + configAnnotation.port(),
+                configAnnotation.user(), configAnnotation.password()));
+      }
+      return new ResourcePollerConfig(SchemaConfig.DEFAULT_POLL_PERIOD,
+          MySQLDbConfig.loadFromEnvironmentVars());
+    }
+  }
 }
diff --git a/sample-operators/mysql-schema/src/main/resources/log4j2.xml b/sample-operators/mysql-schema/src/main/resources/log4j2.xml
index 5ab4735126..01484221f9 100644
--- a/sample-operators/mysql-schema/src/main/resources/log4j2.xml
+++ b/sample-operators/mysql-schema/src/main/resources/log4j2.xml
@@ -6,7 +6,7 @@
         </Console>
     </Appenders>
     <Loggers>
-        <Root level="info">
+        <Root level="debug">
             <AppenderRef ref="Console"/>
         </Root>
     </Loggers>
diff --git a/sample-operators/mysql-schema/src/test/java/io/javaoperatorsdk/operator/sample/MySQLSchemaOperatorE2E.java b/sample-operators/mysql-schema/src/test/java/io/javaoperatorsdk/operator/sample/MySQLSchemaOperatorE2E.java
index 24e257956b..e6aa796656 100644
--- a/sample-operators/mysql-schema/src/test/java/io/javaoperatorsdk/operator/sample/MySQLSchemaOperatorE2E.java
+++ b/sample-operators/mysql-schema/src/test/java/io/javaoperatorsdk/operator/sample/MySQLSchemaOperatorE2E.java
@@ -53,7 +53,7 @@ class MySQLSchemaOperatorE2E {
   boolean isLocal() {
     String deployment = System.getProperty("test.deployment");
     boolean remote = (deployment != null && deployment.equals("remote"));
-    log.info("Running the operator " + (remote ? "remote" : "locally"));
+    log.info("Running the operator " + (remote ? "remotely" : "locally"));
     return !remote;
   }