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 8eb713fa73..43c61319ac 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
@@ -2,7 +2,6 @@
 
 import java.lang.annotation.Annotation;
 import java.time.Duration;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -15,20 +14,18 @@
 import io.javaoperatorsdk.operator.OperatorException;
 import io.javaoperatorsdk.operator.ReconcilerUtils;
 import io.javaoperatorsdk.operator.api.config.dependent.DependentResourceSpec;
-import io.javaoperatorsdk.operator.api.reconciler.*;
+import io.javaoperatorsdk.operator.api.reconciler.Constants;
 import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
+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.processing.dependent.kubernetes.KubernetesDependent;
-import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResource;
-import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResourceConfig;
+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;
 import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEventFilters;
 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.retry.Retry;
 
@@ -158,7 +155,7 @@ public Retry getRetry() {
 
 
   @SuppressWarnings("unchecked")
-  private <T> void configureFromAnnotatedReconciler(T instance) {
+  private void configureFromAnnotatedReconciler(Object instance) {
     if (instance instanceof AnnotationConfigurable) {
       AnnotationConfigurable configurable = (AnnotationConfigurable) instance;
       final Class<? extends Annotation> configurationClass =
@@ -171,6 +168,22 @@ private <T> void configureFromAnnotatedReconciler(T 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() {
@@ -208,11 +221,7 @@ public List<DependentResourceSpec> getDependentResources() {
 
       final var specsMap = new LinkedHashMap<String, DependentResourceSpec>(dependents.length);
       for (Dependent dependent : dependents) {
-        Object config = null;
         final Class<? extends DependentResource> dependentType = dependent.type();
-        if (KubernetesDependentResource.class.isAssignableFrom(dependentType)) {
-          config = createKubernetesResourceConfig(dependentType);
-        }
 
         final var name = getName(dependent, dependentType);
         var spec = specsMap.get(name);
@@ -220,10 +229,16 @@ public List<DependentResourceSpec> getDependentResources() {
           throw new IllegalArgumentException(
               "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(dependentType, config, name,
+        spec = new DependentResourceSpec(dependentResource, name,
             Set.of(dependent.dependsOn()),
             Utils.instantiate(dependent.readyPostcondition(), Condition.class, context),
             Utils.instantiate(dependent.reconcilePrecondition(), Condition.class, context),
@@ -245,52 +260,6 @@ private String getName(Dependent dependent, Class<? extends DependentResource> d
     return name;
   }
 
-  @SuppressWarnings({"rawtypes", "unchecked"})
-  private Object createKubernetesResourceConfig(Class<? extends DependentResource> dependentType) {
-
-    Object config;
-    final var kubeDependent = dependentType.getAnnotation(KubernetesDependent.class);
-
-    var namespaces = 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<?, ? extends HasMetadata> 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(this, dependentType, null);
-      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);
-    }
-
-    config =
-        new KubernetesDependentResourceConfig(namespaces, labelSelector, configuredNS,
-            resourceDiscriminator, onAddFilter,
-            onUpdateFilter, onDeleteFilter, genericFilter);
-
-    return config;
-  }
-
   public static <T> T valueOrDefault(
       ControllerConfiguration controllerConfiguration,
       Function<ControllerConfiguration, T> mapper,
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 00e0b84526..d03cd9f2fa 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
@@ -140,8 +140,9 @@ default ObjectMapper getObjectMapper() {
     return Serialization.jsonMapper();
   }
 
+  @Deprecated(forRemoval = true)
   default DependentResourceFactory dependentResourceFactory() {
-    return new DependentResourceFactory() {};
+    return null;
   }
 
   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 511596db00..c36aa51d2e 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
@@ -6,12 +6,11 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.Set;
-import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.config.dependent.DependentResourceSpec;
-import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResourceConfig;
+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;
@@ -159,6 +158,7 @@ public ControllerConfigurationOverrider<R> withGenericFilter(GenericFilter<R> ge
     return this;
   }
 
+  @SuppressWarnings("unchecked")
   public ControllerConfigurationOverrider<R> replacingNamedDependentResourceConfig(String name,
       Object dependentResourceConfig) {
 
@@ -166,35 +166,31 @@ public ControllerConfigurationOverrider<R> replacingNamedDependentResourceConfig
     if (current == null) {
       throw new IllegalArgumentException("Cannot find a DependentResource named: " + name);
     }
-    replaceConfig(name, dependentResourceConfig, current);
-    return this;
-  }
 
-  private void replaceConfig(String name, Object newConfig, DependentResourceSpec<?, ?> current) {
-    namedDependentResourceSpecs.put(name,
-        new DependentResourceSpec<>(current.getDependentResourceClass(), newConfig, name,
-            current.getDependsOn(), current.getReadyCondition(), current.getReconcileCondition(),
-            current.getDeletePostCondition(), current.getUseEventSourceWithName().orElse(null)));
+    var dependentResource = current.getDependentResource();
+    if (dependentResource instanceof DependentResourceConfigurator) {
+      var configurator = (DependentResourceConfigurator) dependentResource;
+      configurator.configureWith(dependentResourceConfig);
+    }
+
+    return this;
   }
 
-  @SuppressWarnings("unchecked")
   public ControllerConfiguration<R> build() {
-    // propagate namespaces if needed
-    final List<DependentResourceSpec> newDependentSpecs;
     final var hasModifiedNamespaces = !original.getNamespaces().equals(namespaces);
-    newDependentSpecs = namedDependentResourceSpecs.entrySet().stream()
-        .map(drsEntry -> {
-          final var spec = drsEntry.getValue();
-
-          // if the spec has a config and it's a KubernetesDependentResourceConfig, update the
-          // namespaces if needed, otherwise, just return the existing spec
-          final Optional<?> maybeConfig = spec.getDependentResourceConfiguration();
-          return maybeConfig.filter(KubernetesDependentResourceConfig.class::isInstance)
-              .map(KubernetesDependentResourceConfig.class::cast)
-              .filter(Predicate.not(KubernetesDependentResourceConfig::wereNamespacesConfigured))
-              .map(c -> updateSpec(drsEntry.getKey(), spec, c))
-              .orElse(drsEntry.getValue());
-        }).collect(Collectors.toUnmodifiableList());
+    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(),
@@ -215,15 +211,6 @@ public ControllerConfiguration<R> build() {
         newDependentSpecs);
   }
 
-  @SuppressWarnings({"rawtypes", "unchecked"})
-  private DependentResourceSpec<?, ?> updateSpec(String name, DependentResourceSpec spec,
-      KubernetesDependentResourceConfig c) {
-    return new DependentResourceSpec(spec.getDependentResourceClass(),
-        c.setNamespaces(namespaces), name, spec.getDependsOn(), spec.getReadyCondition(),
-        spec.getReconcileCondition(), spec.getDeletePostCondition(),
-        (String) spec.getUseEventSourceWithName().orElse(null));
-  }
-
   public static <R extends HasMetadata> ControllerConfigurationOverrider<R> override(
       ControllerConfiguration<R> original) {
     return new ControllerConfigurationOverrider<>(original);
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/NamespaceChangeable.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/NamespaceChangeable.java
index 4e3cb19ed9..426b179438 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/NamespaceChangeable.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/NamespaceChangeable.java
@@ -16,9 +16,9 @@ public interface NamespaceChangeable {
    */
   void changeNamespaces(Set<String> namespaces);
 
+  @SuppressWarnings("unused")
   default void changeNamespaces(String... namespaces) {
-    changeNamespaces(
-        namespaces != null ? Set.of(namespaces) : DEFAULT_NAMESPACES_SET);
+    changeNamespaces(namespaces != null ? Set.of(namespaces) : DEFAULT_NAMESPACES_SET);
   }
 
   default boolean allowsNamespaceChanges() {
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 98d835af58..fd0f612fa1 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
@@ -111,6 +111,11 @@ public static Class<?> getFirstTypeArgumentFromExtendedClass(Class<?> clazz) {
 
   public static Class<?> getFirstTypeArgumentFromInterface(Class<?> clazz,
       Class<?> expectedImplementedInterface) {
+    return getTypeArgumentFromInterfaceByIndex(clazz, expectedImplementedInterface, 0);
+  }
+
+  public static Class<?> getTypeArgumentFromInterfaceByIndex(Class<?> clazz,
+      Class<?> expectedImplementedInterface, int index) {
     if (expectedImplementedInterface.isAssignableFrom(clazz)) {
       final var genericInterfaces = clazz.getGenericInterfaces();
       Optional<? extends Class<?>> target = Optional.empty();
@@ -122,7 +127,7 @@ public static Class<?> getFirstTypeArgumentFromInterface(Class<?> clazz,
             .map(ParameterizedType.class::cast)
             .findFirst()
             .map(t -> {
-              final Type argument = t.getActualTypeArguments()[0];
+              final Type argument = t.getActualTypeArguments()[index];
               if (argument instanceof Class) {
                 return (Class<?>) argument;
               }
@@ -148,7 +153,7 @@ public static Class<?> getFirstTypeArgumentFromInterface(Class<?> clazz,
       // try the parent
       var parent = clazz.getSuperclass();
       if (!Object.class.equals(parent)) {
-        return getFirstTypeArgumentFromInterface(parent, expectedImplementedInterface);
+        return getTypeArgumentFromInterfaceByIndex(parent, expectedImplementedInterface, index);
       }
     }
     throw new IllegalArgumentException("Couldn't retrieve generic parameter type from "
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 426c1d3325..5d0b8c6b01 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
@@ -4,14 +4,14 @@
 import java.util.Optional;
 import java.util.Set;
 
+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<T extends DependentResource<?, ?>, C> {
+public class DependentResourceSpec<R, P extends HasMetadata, C> {
 
-  private final Class<T> dependentResourceClass;
-
-  private final C dependentResourceConfig;
+  private final DependentResource<R, P> dependentResource;
 
   private final String name;
 
@@ -25,12 +25,11 @@ public class DependentResourceSpec<T extends DependentResource<?, ?>, C> {
 
   private final String useEventSourceWithName;
 
-  public DependentResourceSpec(Class<T> dependentResourceClass, C dependentResourceConfig,
+  public DependentResourceSpec(DependentResource<R, P> dependentResource,
       String name, Set<String> dependsOn, Condition<?, ?> readyCondition,
       Condition<?, ?> reconcileCondition, Condition<?, ?> deletePostCondition,
       String useEventSourceWithName) {
-    this.dependentResourceClass = dependentResourceClass;
-    this.dependentResourceConfig = dependentResourceConfig;
+    this.dependentResource = dependentResource;
     this.name = name;
     this.dependsOn = dependsOn;
     this.readyCondition = readyCondition;
@@ -39,12 +38,28 @@ public DependentResourceSpec(Class<T> dependentResourceClass, C dependentResourc
     this.useEventSourceWithName = useEventSourceWithName;
   }
 
-  public Class<T> getDependentResourceClass() {
-    return dependentResourceClass;
+  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() {
-    return Optional.ofNullable(dependentResourceConfig);
+    if (dependentResource instanceof DependentResourceConfigurator) {
+      var configurator = (DependentResourceConfigurator) dependentResource;
+      return configurator.configuration();
+    }
+    return Optional.empty();
   }
 
   public String getName() {
@@ -54,8 +69,7 @@ public String getName() {
   @Override
   public String toString() {
     return "DependentResourceSpec{ name='" + name +
-        "', type=" + dependentResourceClass.getCanonicalName() +
-        ", config=" + dependentResourceConfig + '}';
+        "', type=" + getDependentResourceClass().getCanonicalName() + '}';
   }
 
   @Override
@@ -66,7 +80,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);
   }
 
@@ -94,6 +108,10 @@ 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 11b6e1f059..139d70b002 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,23 +1,18 @@
 package io.javaoperatorsdk.operator.api.reconciler.dependent;
 
-import java.lang.reflect.InvocationTargetException;
-
+import io.javaoperatorsdk.operator.api.config.Utils;
 import io.javaoperatorsdk.operator.api.config.dependent.DependentResourceSpec;
 
+@Deprecated
+@SuppressWarnings({"rawtypes", "unchecked"})
 public interface DependentResourceFactory {
 
-  default <T extends DependentResource<?, ?>> T createFrom(DependentResourceSpec<T, ?> spec) {
+  default DependentResource createFrom(DependentResourceSpec spec) {
     return createFrom(spec.getDependentResourceClass());
   }
 
-  default <T extends DependentResource<?, ?>> T createFrom(Class<T> dependentResourceClass) {
-    try {
-      return dependentResourceClass.getConstructor().newInstance();
-    } catch (InstantiationException | NoSuchMethodException | IllegalAccessException
-        | InvocationTargetException e) {
-      throw new IllegalArgumentException("Cannot instantiate DependentResource "
-          + dependentResourceClass.getCanonicalName(), e);
-    }
+  default <T extends DependentResource> T createFrom(Class<T> dependentResourceClass) {
+    return (T) Utils.instantiate(dependentResourceClass, DependentResource.class, null);
   }
 
 }
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
new file mode 100644
index 0000000000..d65249b753
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/managed/AnnotationDependentResourceConfigurator.java
@@ -0,0 +1,11 @@
+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/api/reconciler/dependent/managed/DependentResourceConfigurator.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/managed/DependentResourceConfigurator.java
index bbb4f75da9..2b361626aa 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/managed/DependentResourceConfigurator.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/managed/DependentResourceConfigurator.java
@@ -1,5 +1,9 @@
 package io.javaoperatorsdk.operator.api.reconciler.dependent.managed;
 
+import java.util.Optional;
+
 public interface DependentResourceConfigurator<C> {
   void configureWith(C config);
+
+  Optional<C> configuration();
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java
index 070c8f8ffe..72cd25ac5b 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,8 @@
 package io.javaoperatorsdk.operator.processing.dependent.kubernetes;
 
+import java.util.Arrays;
 import java.util.HashMap;
+import java.util.Optional;
 import java.util.Set;
 
 import org.slf4j.Logger;
@@ -10,19 +12,26 @@
 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.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.DependentResourceConfigurator;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.AnnotationDependentResourceConfigurator;
 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;
 
@@ -31,7 +40,7 @@
 public abstract class KubernetesDependentResource<R extends HasMetadata, P extends HasMetadata>
     extends AbstractEventSourceHolderDependentResource<R, P, InformerEventSource<R, P>>
     implements KubernetesClientAware,
-    DependentResourceConfigurator<KubernetesDependentResourceConfig<R>> {
+    AnnotationDependentResourceConfigurator<KubernetesDependent, KubernetesDependentResourceConfig<R>> {
 
   private static final Logger log = LoggerFactory.getLogger(KubernetesDependentResource.class);
 
@@ -39,7 +48,7 @@ public abstract class KubernetesDependentResource<R extends HasMetadata, P exten
   private final Matcher<R, P> matcher;
   private final ResourceUpdatePreProcessor<R> processor;
   private final boolean garbageCollected = this instanceof GarbageCollected;
-  private KubernetesDependentResourceConfig kubernetesDependentResourceConfig;
+  private KubernetesDependentResourceConfig<R> kubernetesDependentResourceConfig;
 
   @SuppressWarnings("unchecked")
   public KubernetesDependentResource(Class<R> resourceType) {
@@ -239,4 +248,49 @@ private void cleanupAfterEventFiltering(ResourceID resourceID) {
     ((InformerEventSource<R, P>) eventSource().orElseThrow())
         .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/kubernetes/KubernetesDependentResourceConfig.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResourceConfig.java
index 5cab02d28c..02a3c8dd78 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResourceConfig.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResourceConfig.java
@@ -2,6 +2,7 @@
 
 import java.util.Set;
 
+import io.javaoperatorsdk.operator.api.config.NamespaceChangeable;
 import io.javaoperatorsdk.operator.api.reconciler.Constants;
 import io.javaoperatorsdk.operator.api.reconciler.ResourceDiscriminator;
 import io.javaoperatorsdk.operator.processing.event.source.filter.GenericFilter;
@@ -11,7 +12,7 @@
 
 import static io.javaoperatorsdk.operator.api.reconciler.Constants.NO_VALUE_SET;
 
-public class KubernetesDependentResourceConfig<R> {
+public class KubernetesDependentResourceConfig<R> implements NamespaceChangeable {
 
   private Set<String> namespaces = Constants.SAME_AS_CONTROLLER_NAMESPACES_SET;
   private String labelSelector = NO_VALUE_SET;
@@ -48,12 +49,6 @@ public KubernetesDependentResourceConfig(Set<String> namespaces, String labelSel
         null, null);
   }
 
-  public KubernetesDependentResourceConfig<R> setNamespaces(Set<String> namespaces) {
-    this.namespacesWereConfigured = true;
-    this.namespaces = namespaces;
-    return this;
-  }
-
   public KubernetesDependentResourceConfig<R> setLabelSelector(String labelSelector) {
     this.labelSelector = labelSelector;
     return this;
@@ -94,4 +89,11 @@ public ResourceDiscriminator getResourceDiscriminator() {
     return resourceDiscriminator;
   }
 
+  @Override
+  public void changeNamespaces(Set<String> namespaces) {
+    if (!wereNamespacesConfigured()) {
+      this.namespacesWereConfigured = true;
+      this.namespaces = namespaces;
+    }
+  }
 }
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 a6e6d27ce3..7a5a7521ba 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
@@ -12,11 +12,9 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.javaoperatorsdk.operator.OperatorException;
-import io.javaoperatorsdk.operator.api.config.ConfigurationServiceProvider;
 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.DependentResourceConfigurator;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.KubernetesClientAware;
 import io.javaoperatorsdk.operator.processing.dependent.workflow.builder.WorkflowBuilder;
 
@@ -73,11 +71,10 @@ public <P extends HasMetadata> Workflow<P> createWorkflow(
     return workflowBuilder.build();
   }
 
-  @SuppressWarnings({"rawtypes", "unchecked"})
+  @SuppressWarnings({"rawtypes"})
   public DependentResource createAndConfigureFrom(DependentResourceSpec spec,
       KubernetesClient client) {
-    final var dependentResource =
-        ConfigurationServiceProvider.instance().dependentResourceFactory().createFrom(spec);
+    final var dependentResource = spec.getDependentResource();
 
     if (dependentResource instanceof KubernetesClientAware) {
       ((KubernetesClientAware) dependentResource).setKubernetesClient(client);
@@ -95,11 +92,6 @@ public DependentResource createAndConfigureFrom(DependentResourceSpec spec,
                     + EventSourceReferencer.class.getSimpleName());
           }
         });
-
-    if (dependentResource instanceof DependentResourceConfigurator) {
-      final var configurator = (DependentResourceConfigurator) dependentResource;
-      spec.getDependentResourceConfiguration().ifPresent(configurator::configureWith);
-    }
     return dependentResource;
   }
 
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverriderTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverriderTest.java
index 878118ba25..0ed9a9f4f6 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverriderTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverriderTest.java
@@ -9,10 +9,7 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.Config;
 import io.fabric8.kubernetes.client.ConfigBuilder;
-import io.javaoperatorsdk.operator.api.config.dependent.DependentResourceSpec;
 import io.javaoperatorsdk.operator.api.monitoring.Metrics;
-import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
-import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResourceFactory;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 
@@ -24,12 +21,6 @@ class ConfigurationServiceOverriderTest {
   private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
   private static final LeaderElectionConfiguration LEADER_ELECTION_CONFIGURATION =
       new LeaderElectionConfiguration("foo", "fooNS");
-  private static final DependentResourceFactory FACTORY = new DependentResourceFactory() {
-    @Override
-    public <T extends DependentResource<?, ?>> T createFrom(DependentResourceSpec<T, ?> spec) {
-      return DependentResourceFactory.super.createFrom(spec);
-    }
-  };
 
   private static final Cloner CLONER = new Cloner() {
     @Override
@@ -86,11 +77,6 @@ public Cloner getResourceCloner() {
         return CLONER;
       }
 
-      @Override
-      public DependentResourceFactory dependentResourceFactory() {
-        return FACTORY;
-      }
-
       @Override
       public Optional<LeaderElectionConfiguration> getLeaderElectionConfiguration() {
         return Optional.of(LEADER_ELECTION_CONFIGURATION);
@@ -123,7 +109,6 @@ public <R extends HasMetadata> R clone(R object) {
         overridden.concurrentReconciliationThreads());
     assertNotEquals(config.getTerminationTimeoutSeconds(),
         overridden.getTerminationTimeoutSeconds());
-    assertNotEquals(config.dependentResourceFactory(), overridden.dependentResourceFactory());
     assertNotEquals(config.getClientConfiguration(), overridden.getClientConfiguration());
     assertNotEquals(config.getExecutorService(), overridden.getExecutorService());
     assertNotEquals(config.getMetrics(), overridden.getMetrics());
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 4dd6938728..f9f75d6796 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
@@ -1,5 +1,6 @@
 package io.javaoperatorsdk.operator.api.config;
 
+import java.util.Optional;
 import java.util.Set;
 
 import org.junit.jupiter.api.Test;
@@ -10,6 +11,7 @@
 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;
@@ -69,7 +71,10 @@ public NamedDependentResource() {
       }
     }
 
-    private static class ExternalDependentResource implements DependentResource<Object, ConfigMap> {
+    private static class ExternalDependentResource implements DependentResource<Object, ConfigMap>,
+        DependentResourceConfigurator<String> {
+
+      private String config = "UNSET";
 
       @Override
       public ReconcileResult<Object> reconcile(ConfigMap primary, Context<ConfigMap> context) {
@@ -80,6 +85,16 @@ public ReconcileResult<Object> reconcile(ConfigMap primary, Context<ConfigMap> c
       public Class<Object> resourceType() {
         return Object.class;
       }
+
+      @Override
+      public void configureWith(String config) {
+        this.config = config;
+      }
+
+      @Override
+      public Optional<String> configuration() {
+        return Optional.of(config);
+      }
     }
   }
 
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 cc4fe9bc48..11200f55d9 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,7 +9,7 @@
 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.DependentResourceConfigurator;
+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;
@@ -97,8 +97,8 @@ void getsFirstTypeArgumentFromInterface() {
         () -> Utils.getFirstTypeArgumentFromInterface(TestKubernetesDependentResource.class,
             DependentResource.class));
 
-    assertThat(Utils.getFirstTypeArgumentFromInterface(TestKubernetesDependentResource.class,
-        DependentResourceConfigurator.class))
+    assertThat(Utils.getTypeArgumentFromInterfaceByIndex(TestKubernetesDependentResource.class,
+        AnnotationDependentResourceConfigurator.class, 1))
         .isEqualTo(KubernetesDependentResourceConfig.class);
   }
 
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 e3b97afb3c..85e4af05eb 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
@@ -10,9 +10,8 @@ public class ManagedWorkflowTestUtils {
 
   @SuppressWarnings("unchecked")
   public static DependentResourceSpec createDRS(String name, String... dependOns) {
-    return new DependentResourceSpec(EmptyTestDependentResource.class,
-        null, name, Set.of(dependOns), null, null, null,
-        null);
+    return new DependentResourceSpec(new EmptyTestDependentResource(), name, Set.of(dependOns),
+        null, null, null, null);
   }
 
 }
diff --git a/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/dependent/SchemaConfig.java b/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/dependent/SchemaConfig.java
new file mode 100644
index 0000000000..c43a2a060f
--- /dev/null
+++ b/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/dependent/SchemaConfig.java
@@ -0,0 +1,23 @@
+package io.javaoperatorsdk.operator.sample.dependent;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.TYPE})
+public @interface SchemaConfig {
+  int DEFAULT_POLL_PERIOD = 500;
+  int DEFAULT_PORT = 3306;
+
+  int pollPeriod() default DEFAULT_POLL_PERIOD;
+
+  String host();
+
+  String user();
+
+  String password();
+
+  int port() default DEFAULT_PORT;
+}
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 c262822d21..8fb8855e41 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
@@ -3,15 +3,19 @@
 import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.SQLException;
-import java.util.*;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.Optional;
+import java.util.Set;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import io.fabric8.kubernetes.api.model.Secret;
+import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.Deleter;
-import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DependentResourceConfigurator;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.AnnotationDependentResourceConfigurator;
 import io.javaoperatorsdk.operator.processing.dependent.Creator;
 import io.javaoperatorsdk.operator.processing.dependent.external.PerResourcePollingDependentResource;
 import io.javaoperatorsdk.operator.sample.MySQLDbConfig;
@@ -23,12 +27,16 @@
 import static io.javaoperatorsdk.operator.sample.dependent.SecretDependentResource.MYSQL_SECRET_USERNAME;
 import static java.lang.String.format;
 
+@SchemaConfig(pollPeriod = 700, host = "127.0.0.1",
+    port = SchemaDependentResource.LOCAL_PORT,
+    user = "root", password = "password") // NOSONAR: password is only used locally, example only
 public class SchemaDependentResource
     extends PerResourcePollingDependentResource<Schema, MySQLSchema>
-    implements DependentResourceConfigurator<ResourcePollerConfig>,
+    implements AnnotationDependentResourceConfigurator<SchemaConfig, ResourcePollerConfig>,
     Creator<Schema, MySQLSchema>, Deleter<MySQLSchema> {
 
   public static final String NAME = "schema";
+  public static final int LOCAL_PORT = 3307;
   private static final Logger log = LoggerFactory.getLogger(SchemaDependentResource.class);
 
   private MySQLDbConfig dbConfig;
@@ -37,12 +45,29 @@ public SchemaDependentResource() {
     super(Schema.class);
   }
 
+  @Override
+  public Optional<ResourcePollerConfig> configuration() {
+    return Optional.of(new ResourcePollerConfig((int) getPollingPeriod(), dbConfig));
+  }
+
   @Override
   public void configureWith(ResourcePollerConfig config) {
     this.dbConfig = config.getMySQLDbConfig();
     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());
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 93a0097b6c..24e257956b 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
@@ -13,12 +13,11 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.api.model.NamespaceBuilder;
 import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
-import io.fabric8.kubernetes.client.DefaultKubernetesClient;
 import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientBuilder;
 import io.javaoperatorsdk.operator.junit.AbstractOperatorExtension;
 import io.javaoperatorsdk.operator.junit.ClusterDeployedOperatorExtension;
 import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension;
-import io.javaoperatorsdk.operator.sample.dependent.ResourcePollerConfig;
 import io.javaoperatorsdk.operator.sample.dependent.SchemaDependentResource;
 
 import static java.util.concurrent.TimeUnit.MINUTES;
@@ -33,13 +32,12 @@ class MySQLSchemaOperatorE2E {
 
   static final Logger log = LoggerFactory.getLogger(MySQLSchemaOperatorE2E.class);
 
-  static final KubernetesClient client = new DefaultKubernetesClient();
+  static final KubernetesClient client = new KubernetesClientBuilder().build();
 
   static final String MY_SQL_NS = "mysql";
 
   private final static List<HasMetadata> infrastructure = new ArrayList<>();
   public static final String TEST_RESOURCE_NAME = "mydb1";
-  public static final Integer LOCAL_PORT = 3307;
 
   static {
     infrastructure.add(
@@ -63,15 +61,10 @@ boolean isLocal() {
   AbstractOperatorExtension operator =
       isLocal()
           ? LocallyRunOperatorExtension.builder()
-              .withReconciler(
-                  new MySQLSchemaReconciler(),
-                  c -> c.replacingNamedDependentResourceConfig(
-                      SchemaDependentResource.NAME,
-                      new ResourcePollerConfig(
-                          700, new MySQLDbConfig("127.0.0.1", LOCAL_PORT.toString(), "root",
-                              "password"))))
+              .withReconciler(new MySQLSchemaReconciler()) // configuration for schema comes from
+                                                           // SchemaDependentResource annotation
               .withInfrastructure(infrastructure)
-              .withPortForward(MY_SQL_NS, "app", "mysql", 3306, LOCAL_PORT)
+              .withPortForward(MY_SQL_NS, "app", "mysql", 3306, SchemaDependentResource.LOCAL_PORT)
               .build()
           : ClusterDeployedOperatorExtension.builder()
               .withOperatorDeployment(client.load(new FileInputStream("k8s/operator.yaml")).get())