diff --git a/docs/documentation/features.md b/docs/documentation/features.md
index 1b9c534f46..67b692f25a 100644
--- a/docs/documentation/features.md
+++ b/docs/documentation/features.md
@@ -458,9 +458,71 @@ following attributes are available in most parts of reconciliation logic and dur
 
 For more information about MDC see this [link](https://www.baeldung.com/mdc-in-log4j-2-logback).
 
+## Dynamically Changing Target Namespaces
+
+A controller can be configured to watch a specific set of namespaces in addition of the 
+namespace in which it is currently deployed or the whole cluster. The framework supports 
+dynamically changing the list of these namespaces while the operator is running. 
+When a reconciler is registered, an instance of 
+[`RegisteredController`](https://github.com/java-operator-sdk/java-operator-sdk/blob/ec37025a15046d8f409c77616110024bf32c3416/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/RegisteredController.java#L5)
+is returned, providing access to the methods allowing users to change watched namespaces as the 
+operator is running.
+
+A typical scenario would probably involve extracting the list of target namespaces from a 
+`ConfigMap` or some other input but this part is out of the scope of the framework since this is 
+use-case specific. For example, reacting to changes to a `ConfigMap` would probably involve 
+registering an associated `Informer` and then calling the `changeNamespaces` method on 
+`RegisteredController`.
+
+```java
+
+  public static void main(String[] args) throws IOException {
+    KubernetesClient client = new DefaultKubernetesClient();
+    Operator operator = new Operator(client);
+    RegisteredController registeredController = operator.register(new WebPageReconciler(client));
+    operator.installShutdownHook();
+    operator.start();
+    
+    // call registeredController further while operator is running
+  }
+
+```
+
+If watched namespaces change for a controller, it might be desirable to propagate these changes to 
+`InformerEventSources` associated with the controller. In order to express this, 
+`InformerEventSource` implementations interested in following such changes need to be 
+configured appropriately so that the `followControllerNamespaceChanges` method returns `true`:
+
+```java
+
+@ControllerConfiguration
+public class MyReconciler
+        implements Reconciler<TestCustomResource>, EventSourceInitializer<TestCustomResource>{
+
+ @Override
+  public Map<String, EventSource> prepareEventSources(
+      EventSourceContext<ChangeNamespaceTestCustomResource> context) {
+
+    InformerEventSource<ConfigMap, TestCustomResource> configMapES =
+        new InformerEventSource<>(InformerConfiguration.from(ConfigMap.class)
+            .withNamespacesInheritedFromController(context)
+            .build(), context);
+
+    return EventSourceInitializer.nameEventSources(configMapES);
+  }
+  
+}
+```
+
+As seen in the above code snippet, the informer will have the initial namespaces inherited from controller, but
+also will adjust the target namespaces if it changes for the controller.
+
+See also the [integration test](https://github.com/java-operator-sdk/java-operator-sdk/blob/ec37025a15046d8f409c77616110024bf32c3416/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/changenamespace/ChangeNamespaceTestReconciler.java)
+for this feature.
+
 ## Monitoring with Micrometer
 
-## Automatic generation of CRDs
+## Automatic Generation of CRDs
 
 Note that this is feature of [Fabric8 Kubernetes Client](https://github.com/fabric8io/kubernetes-client) not the JOSDK.
 But it's worth to mention here.
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/Operator.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/Operator.java
index 9765ede762..21b1216ced 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/Operator.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/Operator.java
@@ -129,11 +129,11 @@ public void stop() throws OperatorException {
    * @param <R> the {@code CustomResource} type associated with the reconciler
    * @throws OperatorException if a problem occurred during the registration process
    */
-  public <R extends HasMetadata> void register(Reconciler<R> reconciler)
+  public <R extends HasMetadata> RegisteredController register(Reconciler<R> reconciler)
       throws OperatorException {
     final var controllerConfiguration =
         ConfigurationServiceProvider.instance().getConfigurationFor(reconciler);
-    register(reconciler, controllerConfiguration);
+    return register(reconciler, controllerConfiguration);
   }
 
   /**
@@ -148,7 +148,7 @@ public <R extends HasMetadata> void register(Reconciler<R> reconciler)
    * @param <R> the {@code HasMetadata} type associated with the reconciler
    * @throws OperatorException if a problem occurred during the registration process
    */
-  public <R extends HasMetadata> void register(Reconciler<R> reconciler,
+  public <R extends HasMetadata> RegisteredController register(Reconciler<R> reconciler,
       ControllerConfiguration<R> configuration)
       throws OperatorException {
 
@@ -173,6 +173,7 @@ public <R extends HasMetadata> void register(Reconciler<R> reconciler,
         configuration.getName(),
         configuration.getResourceClass(),
         watchedNS);
+    return controller;
   }
 
   /**
@@ -182,13 +183,13 @@ public <R extends HasMetadata> void register(Reconciler<R> reconciler,
    * @param configOverrider consumer to use to change config values
    * @param <R> the {@code HasMetadata} type associated with the reconciler
    */
-  public <R extends HasMetadata> void register(Reconciler<R> reconciler,
+  public <R extends HasMetadata> RegisteredController register(Reconciler<R> reconciler,
       Consumer<ControllerConfigurationOverrider<R>> configOverrider) {
     final var controllerConfiguration =
         ConfigurationServiceProvider.instance().getConfigurationFor(reconciler);
     var configToOverride = ControllerConfigurationOverrider.override(controllerConfiguration);
     configOverrider.accept(configToOverride);
-    register(reconciler, configToOverride.build());
+    return register(reconciler, configToOverride.build());
   }
 
   static class ControllerManager implements LifecycleAware {
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/RegisteredController.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/RegisteredController.java
new file mode 100644
index 0000000000..f1fa1f37c4
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/RegisteredController.java
@@ -0,0 +1,6 @@
+package io.javaoperatorsdk.operator;
+
+import io.javaoperatorsdk.operator.api.config.NamespaceChangeable;
+
+public interface RegisteredController extends NamespaceChangeable {
+}
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
new file mode 100644
index 0000000000..d03f59a8d0
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/NamespaceChangeable.java
@@ -0,0 +1,26 @@
+package io.javaoperatorsdk.operator.api.config;
+
+import java.util.Set;
+
+public interface NamespaceChangeable {
+
+  /**
+   * If the controller and possibly registered
+   * {@link io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource}
+   * watches a set of namespaces this set can be adjusted dynamically, this when the operator is
+   * running.
+   *
+   * @param namespaces target namespaces to watch
+   */
+  void changeNamespaces(Set<String> namespaces);
+
+  default void changeNamespaces(String... namespaces) {
+    changeNamespaces(
+        namespaces != null ? Set.of(namespaces) : ResourceConfiguration.DEFAULT_NAMESPACES);
+  }
+
+  default boolean allowsNamespaceChanges() {
+    return true;
+  }
+
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ResourceConfiguration.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ResourceConfiguration.java
index fe85476735..833063af93 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ResourceConfiguration.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ResourceConfiguration.java
@@ -10,8 +10,8 @@
 
 public interface ResourceConfiguration<R extends HasMetadata> {
 
-  Set<String> DEFAULT_NAMESPACES = Set.of(Constants.WATCH_ALL_NAMESPACES);
-  Set<String> CURRENT_NAMESPACE_ONLY = Set.of(Constants.WATCH_CURRENT_NAMESPACE);
+  Set<String> DEFAULT_NAMESPACES = Collections.singleton(Constants.WATCH_ALL_NAMESPACES);
+  Set<String> CURRENT_NAMESPACE_ONLY = Collections.singleton(Constants.WATCH_CURRENT_NAMESPACE);
 
   default String getResourceTypeName() {
     return ReconcilerUtils.getResourceTypeName(getResourceClass());
@@ -64,7 +64,6 @@ static void failIfNotValid(Set<String> namespaces) {
         return;
       }
     }
-
     throw new IllegalArgumentException(
         "Must specify namespaces. To watch all namespaces, use only '"
             + Constants.WATCH_ALL_NAMESPACES
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerConfiguration.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerConfiguration.java
index a2105cac11..b3388f969a 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerConfiguration.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerConfiguration.java
@@ -1,16 +1,15 @@
 package io.javaoperatorsdk.operator.api.config.informer;
 
-import java.util.Collections;
 import java.util.Objects;
 import java.util.Set;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.config.DefaultResourceConfiguration;
 import io.javaoperatorsdk.operator.api.config.ResourceConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
 import io.javaoperatorsdk.operator.processing.event.source.SecondaryToPrimaryMapper;
 import io.javaoperatorsdk.operator.processing.event.source.informer.Mappers;
 
-@SuppressWarnings("rawtypes")
 public interface InformerConfiguration<R extends HasMetadata>
     extends ResourceConfiguration<R> {
 
@@ -18,17 +17,22 @@ class DefaultInformerConfiguration<R extends HasMetadata> extends
       DefaultResourceConfiguration<R> implements InformerConfiguration<R> {
 
     private final SecondaryToPrimaryMapper<R> secondaryToPrimaryMapper;
+    private final boolean followControllerNamespaceChanges;
 
     protected DefaultInformerConfiguration(String labelSelector,
         Class<R> resourceClass,
         SecondaryToPrimaryMapper<R> secondaryToPrimaryMapper,
-        Set<String> namespaces) {
+        Set<String> namespaces, boolean followControllerNamespaceChanges) {
       super(labelSelector, resourceClass, namespaces);
+      this.followControllerNamespaceChanges = followControllerNamespaceChanges;
       this.secondaryToPrimaryMapper =
           Objects.requireNonNullElse(secondaryToPrimaryMapper,
               Mappers.fromOwnerReference());
     }
 
+    public boolean followControllerNamespaceChanges() {
+      return followControllerNamespaceChanges;
+    }
 
     public SecondaryToPrimaryMapper<R> getSecondaryToPrimaryMapper() {
       return secondaryToPrimaryMapper;
@@ -36,6 +40,14 @@ public SecondaryToPrimaryMapper<R> getSecondaryToPrimaryMapper() {
 
   }
 
+  /**
+   * Used in case the watched namespaces are changed dynamically, thus when operator is running (See
+   * {@link io.javaoperatorsdk.operator.RegisteredController}). If true, changing the target
+   * namespaces of a controller would result to change target namespaces for the
+   * InformerEventSource.
+   */
+  boolean followControllerNamespaceChanges();
+
   SecondaryToPrimaryMapper<R> getSecondaryToPrimaryMapper();
 
   @SuppressWarnings("unused")
@@ -45,6 +57,7 @@ class InformerConfigurationBuilder<R extends HasMetadata> {
     private Set<String> namespaces;
     private String labelSelector;
     private final Class<R> resourceClass;
+    private boolean inheritControllerNamespacesOnChange = false;
 
     private InformerConfigurationBuilder(Class<R> resourceClass) {
       this.resourceClass = resourceClass;
@@ -57,15 +70,61 @@ public InformerConfigurationBuilder<R> withSecondaryToPrimaryMapper(
     }
 
     public InformerConfigurationBuilder<R> withNamespaces(String... namespaces) {
-      this.namespaces = namespaces != null ? Set.of(namespaces) : Collections.emptySet();
-      return this;
+      return withNamespaces(
+          namespaces != null ? Set.of(namespaces) : ResourceConfiguration.DEFAULT_NAMESPACES);
     }
 
     public InformerConfigurationBuilder<R> withNamespaces(Set<String> namespaces) {
-      this.namespaces = namespaces != null ? namespaces : Collections.emptySet();
+      return withNamespaces(namespaces, false);
+    }
+
+    /**
+     * Sets the initial set of namespaces to watch (typically extracted from the parent
+     * {@link io.javaoperatorsdk.operator.processing.Controller}'s configuration), specifying
+     * whether changes made to the parent controller configured namespaces should be tracked or not.
+     *
+     * @param namespaces the initial set of namespaces to watch
+     * @param followChanges {@code true} to follow the changes made to the parent controller
+     *        namespaces, {@code false} otherwise
+     * @return the builder instance so that calls can be chained fluently
+     */
+    public InformerConfigurationBuilder<R> withNamespaces(Set<String> namespaces,
+        boolean followChanges) {
+      this.namespaces = namespaces != null ? namespaces : ResourceConfiguration.DEFAULT_NAMESPACES;
+      this.inheritControllerNamespacesOnChange = true;
+      return this;
+    }
+
+    /**
+     * Configures the informer to watch and track the same namespaces as the parent
+     * {@link io.javaoperatorsdk.operator.processing.Controller}, meaning that the informer will be
+     * restarted to watch the new namespaces if the parent controller's namespace configuration
+     * changes.
+     *
+     * @param context {@link EventSourceContext} from which the parent
+     *        {@link io.javaoperatorsdk.operator.processing.Controller}'s configuration is retrieved
+     * @param <P> the primary resource type associated with the parent controller
+     * @return the builder instance so that calls can be chained fluently
+     */
+    public <P extends HasMetadata> InformerConfigurationBuilder<R> withNamespacesInheritedFromController(
+        EventSourceContext<P> context) {
+      namespaces = context.getControllerConfiguration().getEffectiveNamespaces();
+      this.inheritControllerNamespacesOnChange = true;
       return this;
     }
 
+    /**
+     * Whether or not the associated informer should track changes made to the parent
+     * {@link io.javaoperatorsdk.operator.processing.Controller}'s namespaces configuration.
+     *
+     * @param followChanges {@code true} to reconfigure the associated informer when the parent
+     *        controller's namespaces are reconfigured, {@code false} otherwise
+     * @return the builder instance so that calls can be chained fluently
+     */
+    public InformerConfigurationBuilder<R> followNamespaceChanges(boolean followChanges) {
+      this.inheritControllerNamespacesOnChange = followChanges;
+      return this;
+    }
 
     public InformerConfigurationBuilder<R> withLabelSelector(String labelSelector) {
       this.labelSelector = labelSelector;
@@ -75,7 +134,7 @@ public InformerConfigurationBuilder<R> withLabelSelector(String labelSelector) {
     public InformerConfiguration<R> build() {
       return new DefaultInformerConfiguration<>(labelSelector, resourceClass,
           secondaryToPrimaryMapper,
-          namespaces);
+          namespaces, inheritControllerNamespacesOnChange);
     }
   }
 
@@ -84,12 +143,19 @@ static <R extends HasMetadata> InformerConfigurationBuilder<R> from(
     return new InformerConfigurationBuilder<>(resourceClass);
   }
 
-
+  /**
+   * Creates a configuration builder that inherits namespaces from the controller and follows
+   * namespaces changes.
+   *
+   * @param resourceClass secondary resource class
+   * @param eventSourceContext of the initializer
+   * @return builder
+   * @param <R> secondary resource type
+   */
   static <R extends HasMetadata> InformerConfigurationBuilder<R> from(
-      InformerConfiguration<R> configuration) {
-    return new InformerConfigurationBuilder<R>(configuration.getResourceClass())
-        .withNamespaces(configuration.getNamespaces())
-        .withLabelSelector(configuration.getLabelSelector())
-        .withSecondaryToPrimaryMapper(configuration.getSecondaryToPrimaryMapper());
+      Class<R> resourceClass, EventSourceContext<?> eventSourceContext) {
+    return new InformerConfigurationBuilder<>(resourceClass)
+        .withNamespacesInheritedFromController(eventSourceContext);
   }
+
 }
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 63b3c85a9c..7723d19007 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
@@ -1,9 +1,6 @@
 package io.javaoperatorsdk.operator.processing;
 
-import java.util.ArrayList;
-import java.util.LinkedHashMap;
-import java.util.Map;
-import java.util.Optional;
+import java.util.*;
 import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
@@ -16,24 +13,13 @@
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.fabric8.kubernetes.client.dsl.MixedOperation;
 import io.fabric8.kubernetes.client.dsl.Resource;
-import io.javaoperatorsdk.operator.AggregatedOperatorException;
-import io.javaoperatorsdk.operator.CustomResourceUtils;
-import io.javaoperatorsdk.operator.MissingCRDException;
-import io.javaoperatorsdk.operator.OperatorException;
+import io.javaoperatorsdk.operator.*;
 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.monitoring.Metrics;
 import io.javaoperatorsdk.operator.api.monitoring.Metrics.ControllerExecution;
-import io.javaoperatorsdk.operator.api.reconciler.Cleaner;
-import io.javaoperatorsdk.operator.api.reconciler.Context;
-import io.javaoperatorsdk.operator.api.reconciler.ContextInitializer;
-import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
-import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
-import io.javaoperatorsdk.operator.api.reconciler.EventSourceInitializer;
-import io.javaoperatorsdk.operator.api.reconciler.Ignore;
-import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
-import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.api.reconciler.*;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.Deleter;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.EventSourceProvider;
@@ -42,10 +28,12 @@
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.ManagedDependentResourceException;
 import io.javaoperatorsdk.operator.processing.event.EventSourceManager;
 
+import static io.javaoperatorsdk.operator.api.reconciler.Constants.WATCH_CURRENT_NAMESPACE;
+
 @SuppressWarnings({"unchecked", "rawtypes"})
 @Ignore
 public class Controller<P extends HasMetadata>
-    implements Reconciler<P>, Cleaner<P>, LifecycleAware {
+    implements Reconciler<P>, Cleaner<P>, LifecycleAware, RegisteredController {
 
   private static final Logger log = LoggerFactory.getLogger(Controller.class);
 
@@ -307,29 +295,14 @@ public void start() throws OperatorException {
 
     // fail early if we're missing the current namespace information
     failOnMissingCurrentNS();
-
     try {
       // check that the custom resource is known by the cluster if configured that way
-      final CustomResourceDefinition crd; // todo: check proper CRD spec version based on config
-      if (ConfigurationServiceProvider.instance().checkCRDAndValidateLocalModel()
-          && CustomResource.class.isAssignableFrom(resClass)) {
-        crd = kubernetesClient.apiextensions().v1().customResourceDefinitions().withName(crdName)
-            .get();
-        if (crd == null) {
-          throwMissingCRDException(crdName, specVersion, controllerName);
-        }
-
-        // Apply validations that are not handled by fabric8
-        CustomResourceUtils.assertCustomResource(resClass, crd);
-      }
-
+      validateCRDWithLocalModelIfRequired(resClass, controllerName, crdName, specVersion);
       final var context = new EventSourceContext<>(
           eventSourceManager.getControllerResourceEventSource(), configuration, kubernetesClient);
 
       initAndRegisterEventSources(context);
-
       eventSourceManager.start();
-
       log.info("'{}' controller started, pending event sources initialization", controllerName);
     } catch (MissingCRDException e) {
       stop();
@@ -337,6 +310,29 @@ public void start() throws OperatorException {
     }
   }
 
+  private void validateCRDWithLocalModelIfRequired(Class<P> resClass, String controllerName,
+      String crdName, String specVersion) {
+    final CustomResourceDefinition crd;
+    if (ConfigurationServiceProvider.instance().checkCRDAndValidateLocalModel()
+        && CustomResource.class.isAssignableFrom(resClass)) {
+      crd = kubernetesClient.apiextensions().v1().customResourceDefinitions().withName(crdName)
+          .get();
+      if (crd == null) {
+        throwMissingCRDException(crdName, specVersion, controllerName);
+      }
+      // Apply validations that are not handled by fabric8
+      CustomResourceUtils.assertCustomResource(resClass, crd);
+    }
+  }
+
+  public void changeNamespaces(Set<String> namespaces) {
+    if (namespaces.contains(Constants.WATCH_ALL_NAMESPACES)
+        || namespaces.contains(WATCH_CURRENT_NAMESPACE)) {
+      throw new OperatorException("Unexpected value in target namespaces: " + namespaces);
+    }
+    eventSourceManager.changeNamespaces(namespaces);
+  }
+
   private void throwMissingCRDException(String crdName, String specVersion, String controllerName) {
     throw new MissingCRDException(
         crdName,
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependent.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependent.java
index 8ce04d752e..bda5dcb1dd 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependent.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependent.java
@@ -10,9 +10,9 @@
 @Retention(RetentionPolicy.RUNTIME)
 @Target({ElementType.TYPE})
 public @interface KubernetesDependent {
-  String SAME_AS_PARENT = "JOSDK_SAME_AS_PARENT";
+  String SAME_AS_CONTROLLER = "JOSDK_SAME_AS_CONTROLLER";
 
-  String[] DEFAULT_NAMESPACES = {SAME_AS_PARENT};
+  String[] DEFAULT_NAMESPACES = {SAME_AS_CONTROLLER};
 
   /**
    * Specified which namespaces this Controller monitors for custom resources events. If no
@@ -21,7 +21,7 @@
    *
    * @return the list of namespaces this controller monitors
    */
-  String[] namespaces() default {SAME_AS_PARENT};
+  String[] namespaces() default {SAME_AS_CONTROLLER};
 
   /**
    * Optional label selector used to identify the set of custom resources the controller will acc
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 ca0b94b230..4196eb0e7e 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
@@ -49,20 +49,21 @@ public KubernetesDependentResource(Class<R> resourceType) {
 
   @Override
   public void configureWith(KubernetesDependentResourceConfig config) {
-    configureWith(config.labelSelector(), config.namespaces());
+    configureWith(config.labelSelector(), config.namespaces(), !config.wereNamespacesConfigured());
   }
 
   @SuppressWarnings("unchecked")
-  private void configureWith(String labelSelector, Set<String> namespaces) {
+  private void configureWith(String labelSelector, Set<String> namespaces,
+      boolean inheritNamespacesOnChange) {
     final SecondaryToPrimaryMapper<R> primaryResourcesRetriever =
         (this instanceof SecondaryToPrimaryMapper) ? (SecondaryToPrimaryMapper<R>) this
             : Mappers.fromOwnerReference();
-    InformerConfiguration<R> ic =
-        InformerConfiguration.from(resourceType())
-            .withLabelSelector(labelSelector)
-            .withNamespaces(namespaces)
-            .withSecondaryToPrimaryMapper(primaryResourcesRetriever)
-            .build();
+    var ic = InformerConfiguration.from(resourceType())
+        .withLabelSelector(labelSelector)
+        .withSecondaryToPrimaryMapper(primaryResourcesRetriever)
+        .withNamespaces(namespaces, inheritNamespacesOnChange)
+        .build();
+
     configureWith(new InformerEventSource<>(ic, client));
   }
 
@@ -78,11 +79,9 @@ public void configureWith(InformerEventSource<R, P> informerEventSource) {
 
   protected R handleCreate(R desired, P primary, Context<P> context) {
     ResourceID resourceID = ResourceID.fromResource(desired);
-    R created = null;
     try {
       prepareEventFiltering(desired, resourceID);
-      created = super.handleCreate(desired, primary, context);
-      return created;
+      return super.handleCreate(desired, primary, context);
     } catch (RuntimeException e) {
       cleanupAfterEventFiltering(resourceID);
       throw e;
@@ -91,11 +90,9 @@ protected R handleCreate(R desired, P primary, Context<P> context) {
 
   protected R handleUpdate(R actual, R desired, P primary, Context<P> context) {
     ResourceID resourceID = ResourceID.fromResource(desired);
-    R updated = null;
     try {
       prepareEventFiltering(desired, resourceID);
-      updated = super.handleUpdate(actual, desired, primary, context);
-      return updated;
+      return super.handleUpdate(actual, desired, primary, context);
     } catch (RuntimeException e) {
       cleanupAfterEventFiltering(resourceID);
       throw e;
@@ -139,7 +136,7 @@ protected NonNamespaceOperation<R, KubernetesResourceList<R>, Resource<R>> prepa
 
   @Override
   protected InformerEventSource<R, P> createEventSource(EventSourceContext<P> context) {
-    configureWith(null, context.getControllerConfiguration().getNamespaces());
+    configureWith(null, context.getControllerConfiguration().getNamespaces(), true);
     log.warn("Using default configuration for " + resourceType().getSimpleName()
         + " KubernetesDependentResource, call configureWith to provide configuration");
     return eventSource();
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java
index 9cdb73b5aa..a7ec8c8494 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java
@@ -98,10 +98,6 @@ private EventProcessor(
     this.eventSourceManager = eventSourceManager;
   }
 
-  EventMarker getEventMarker() {
-    return eventMarker;
-  }
-
   @Override
   public void handleEvent(Event event) {
     lock.lock();
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSourceManager.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSourceManager.java
index 1ccc0ad2d9..3990c8c300 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSourceManager.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSourceManager.java
@@ -4,7 +4,6 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.Set;
-import java.util.concurrent.locks.ReentrantLock;
 import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
@@ -13,6 +12,7 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.MissingCRDException;
 import io.javaoperatorsdk.operator.OperatorException;
+import io.javaoperatorsdk.operator.api.config.NamespaceChangeable;
 import io.javaoperatorsdk.operator.api.reconciler.EventSourceInitializer;
 import io.javaoperatorsdk.operator.processing.Controller;
 import io.javaoperatorsdk.operator.processing.LifecycleAware;
@@ -27,18 +27,16 @@ public class EventSourceManager<R extends HasMetadata> implements LifecycleAware
 
   private static final Logger log = LoggerFactory.getLogger(EventSourceManager.class);
 
-  private final ReentrantLock lock = new ReentrantLock();
-  private final EventSources<R> eventSources = new EventSources<>();
+  private final EventSources<R> eventSources;
   private final EventProcessor<R> eventProcessor;
   private final Controller<R> controller;
 
-  EventSourceManager(EventProcessor<R> eventProcessor) {
-    this.eventProcessor = eventProcessor;
-    controller = null;
-    registerEventSource(eventSources.retryEventSource());
+  public EventSourceManager(Controller<R> controller) {
+    this(controller, new EventSources<>());
   }
 
-  public EventSourceManager(Controller<R> controller) {
+  EventSourceManager(Controller<R> controller, EventSources<R> eventSources) {
+    this.eventSources = eventSources;
     this.controller = controller;
     // controller event source needs to be available before we create the event processor
     final var controllerEventSource = eventSources.initControllerEventSource(controller);
@@ -56,29 +54,24 @@ public EventSourceManager(Controller<R> controller) {
    * caches propagated - although for non k8s related event sources this behavior might be different
    * (see
    * {@link io.javaoperatorsdk.operator.processing.event.source.polling.PerResourcePollingEventSource}).
-   *
+   * <p>
    * Now the event sources are also started sequentially, mainly because others might depend on
    * {@link ControllerResourceEventSource} , which is started first.
    */
   @Override
-  public void start() {
-    lock.lock();
-    try {
-      for (var eventSource : eventSources) {
-        try {
-          logEventSourceEvent(eventSource, "Starting");
-          eventSource.start();
-          logEventSourceEvent(eventSource, "Started");
-        } catch (MissingCRDException e) {
-          throw e; // leave untouched
-        } catch (Exception e) {
-          throw new OperatorException("Couldn't start source " + eventSource.name(), e);
-        }
+  public synchronized void start() {
+    for (var eventSource : eventSources) {
+      try {
+        logEventSourceEvent(eventSource, "Starting");
+        eventSource.start();
+        logEventSourceEvent(eventSource, "Started");
+      } catch (MissingCRDException e) {
+        throw e; // leave untouched
+      } catch (Exception e) {
+        throw new OperatorException("Couldn't start source " + eventSource.name(), e);
       }
-      eventProcessor.start();
-    } finally {
-      lock.unlock();
     }
+    eventProcessor.start();
   }
 
   @SuppressWarnings("rawtypes")
@@ -95,22 +88,17 @@ private void logEventSourceEvent(NamedEventSource eventSource, String event) {
   }
 
   @Override
-  public void stop() {
-    lock.lock();
-    try {
-      for (var eventSource : eventSources) {
-        try {
-          logEventSourceEvent(eventSource, "Stopping");
-          eventSource.stop();
-          logEventSourceEvent(eventSource, "Stopped");
-        } catch (Exception e) {
-          log.warn("Error closing {} -> {}", eventSource.name(), e);
-        }
+  public synchronized void stop() {
+    for (var eventSource : eventSources) {
+      try {
+        logEventSourceEvent(eventSource, "Stopping");
+        eventSource.stop();
+        logEventSourceEvent(eventSource, "Stopped");
+      } catch (Exception e) {
+        log.warn("Error closing {} -> {}", eventSource.name(), e);
       }
-      eventSources.clear();
-    } finally {
-      lock.unlock();
     }
+    eventSources.clear();
     eventProcessor.stop();
   }
 
@@ -118,10 +106,9 @@ public final void registerEventSource(EventSource eventSource) throws OperatorEx
     registerEventSource(null, eventSource);
   }
 
-  public final void registerEventSource(String name, EventSource eventSource)
+  public final synchronized void registerEventSource(String name, EventSource eventSource)
       throws OperatorException {
     Objects.requireNonNull(eventSource, "EventSource must not be null");
-    lock.lock();
     try {
       if (name == null || name.isBlank()) {
         name = EventSourceInitializer.generateNameFor(eventSource);
@@ -133,8 +120,6 @@ public final void registerEventSource(String name, EventSource eventSource)
     } catch (Exception e) {
       throw new OperatorException("Couldn't register event source: " + name + " for "
           + controller.getConfiguration().getName() + " controller`", e);
-    } finally {
-      lock.unlock();
     }
   }
 
@@ -158,11 +143,22 @@ public void broadcastOnResourceEvent(ResourceAction action, R resource, R oldRes
     }
   }
 
+  public void changeNamespaces(Set<String> namespaces) {
+    eventProcessor.stop();
+    eventSources
+        .allEventSources()
+        .filter(NamespaceChangeable.class::isInstance)
+        .map(NamespaceChangeable.class::cast)
+        .filter(NamespaceChangeable::allowsNamespaceChanges)
+        .forEach(ies -> ies.changeNamespaces(namespaces));
+    eventProcessor.start();
+  }
+
   EventHandler getEventHandler() {
     return eventProcessor;
   }
 
-  Set<EventSource> getRegisteredEventSources() {
+  public Set<EventSource> getRegisteredEventSources() {
     return eventSources.flatMappedSources()
         .map(NamedEventSource::original)
         .collect(Collectors.toCollection(LinkedHashSet::new));
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSources.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSources.java
index 1d2d343831..674648bb0f 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSources.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSources.java
@@ -45,6 +45,12 @@ Stream<NamedEventSource> flatMappedSources() {
         .map(esEntry -> new NamedEventSource(esEntry.getValue(), esEntry.getKey())));
   }
 
+  Stream<EventSource> allEventSources() {
+    return Stream.concat(
+        Stream.of(retryEventSource(), controllerResourceEventSource()).filter(Objects::nonNull),
+        sources.values().stream().flatMap(c -> c.values().stream()));
+  }
+
   public void clear() {
     sources.clear();
   }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerResourceEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerResourceEventSource.java
index 0c0f0c6b8e..233cecb644 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerResourceEventSource.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerResourceEventSource.java
@@ -24,7 +24,6 @@ public class ControllerResourceEventSource<T extends HasMetadata>
     extends ManagedInformerEventSource<T, T, ControllerConfiguration<T>>
     implements ResourceEventHandler<T> {
 
-  public static final String ANY_NAMESPACE_MAP_KEY = "anyNamespace";
   private static final Logger log = LoggerFactory.getLogger(ControllerResourceEventSource.class);
 
   private final Controller<T> controller;
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSource.java
index 5d1d12875f..47fe4abcf3 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSource.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSource.java
@@ -266,4 +266,8 @@ public synchronized void cleanupOnCreateOrUpdateEventFiltering(ResourceID resour
     eventRecorder.stopEventRecording(resourceID);
   }
 
+  @Override
+  public boolean allowsNamespaceChanges() {
+    return getConfiguration().followControllerNamespaceChanges();
+  }
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManager.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManager.java
index 514caa2a2e..da4631a24f 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManager.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManager.java
@@ -29,11 +29,15 @@
 public class InformerManager<T extends HasMetadata, C extends ResourceConfiguration<T>>
     implements LifecycleAware, IndexerResourceCache<T>, UpdatableCache<T> {
 
-  private static final String ANY_NAMESPACE_MAP_KEY = "anyNamespace";
+  private static final String ALL_NAMESPACES_MAP_KEY = "allNamespaces";
   private static final Logger log = LoggerFactory.getLogger(InformerManager.class);
 
   private final Map<String, InformerWrapper<T>> sources = new ConcurrentHashMap<>();
   private Cloner cloner;
+  private C configuration;
+  private MixedOperation<T, KubernetesResourceList<T>, Resource<T>> client;
+  private ResourceEventHandler<T> eventHandler;
+  private final Map<String, Function<T, List<String>>> indexers = new HashMap<>();
 
   @Override
   public void start() throws OperatorException {
@@ -43,6 +47,9 @@ public void start() throws OperatorException {
   void initSources(MixedOperation<T, KubernetesResourceList<T>, Resource<T>> client,
       C configuration, ResourceEventHandler<T> eventHandler) {
     cloner = ConfigurationServiceProvider.instance().getResourceCloner();
+    this.configuration = configuration;
+    this.client = client;
+    this.eventHandler = eventHandler;
 
     final var targetNamespaces = configuration.getEffectiveNamespaces();
     final var labelSelector = configuration.getLabelSelector();
@@ -51,7 +58,7 @@ void initSources(MixedOperation<T, KubernetesResourceList<T>, Resource<T>> clien
       final var filteredBySelectorClient =
           client.inAnyNamespace().withLabelSelector(labelSelector);
       final var source =
-          createEventSource(filteredBySelectorClient, eventHandler, ANY_NAMESPACE_MAP_KEY);
+          createEventSource(filteredBySelectorClient, eventHandler, ALL_NAMESPACES_MAP_KEY);
       log.debug("Registered {} -> {} for any namespace", this, source);
     } else {
       targetNamespaces.forEach(
@@ -65,6 +72,27 @@ void initSources(MixedOperation<T, KubernetesResourceList<T>, Resource<T>> clien
     }
   }
 
+  public void changeNamespaces(Set<String> namespaces) {
+    var sourcesToRemove = sources.keySet().stream()
+        .filter(k -> !namespaces.contains(k)).collect(Collectors.toSet());
+    log.debug("Stopped informer {} for namespaces: {}", this, sourcesToRemove);
+    sourcesToRemove.forEach(k -> sources.remove(k).stop());
+
+    namespaces.forEach(ns -> {
+      if (!sources.containsKey(ns)) {
+        final var source =
+            createEventSource(
+                client.inNamespace(ns).withLabelSelector(configuration.getLabelSelector()),
+                eventHandler, ns);
+        source.addIndexers(this.indexers);
+        source.start();
+        log.debug("Registered new {} -> {} for namespace: {}", this, source,
+            ns);
+      }
+    });
+  }
+
+
 
   private InformerWrapper<T> createEventSource(
       FilterWatchListDeletable<T, KubernetesResourceList<T>> filteredBySelectorClient,
@@ -98,7 +126,7 @@ public Stream<T> list(Predicate<T> predicate) {
   @Override
   public Stream<T> list(String namespace, Predicate<T> predicate) {
     if (isWatchingAllNamespaces()) {
-      return getSource(ANY_NAMESPACE_MAP_KEY)
+      return getSource(ALL_NAMESPACES_MAP_KEY)
           .map(source -> source.list(namespace, predicate))
           .orElse(Stream.empty());
     } else {
@@ -110,7 +138,7 @@ public Stream<T> list(String namespace, Predicate<T> predicate) {
 
   @Override
   public Optional<T> get(ResourceID resourceID) {
-    return getSource(resourceID.getNamespace().orElse(ANY_NAMESPACE_MAP_KEY))
+    return getSource(resourceID.getNamespace().orElse(ALL_NAMESPACES_MAP_KEY))
         .flatMap(source -> source.get(resourceID))
         .map(cloner::clone);
   }
@@ -121,24 +149,24 @@ public Stream<ResourceID> keys() {
   }
 
   private boolean isWatchingAllNamespaces() {
-    return sources.containsKey(ANY_NAMESPACE_MAP_KEY);
+    return sources.containsKey(ALL_NAMESPACES_MAP_KEY);
   }
 
   private Optional<InformerWrapper<T>> getSource(String namespace) {
-    namespace = isWatchingAllNamespaces() || namespace == null ? ANY_NAMESPACE_MAP_KEY : namespace;
+    namespace = isWatchingAllNamespaces() || namespace == null ? ALL_NAMESPACES_MAP_KEY : namespace;
     return Optional.ofNullable(sources.get(namespace));
   }
 
   @Override
   public T remove(ResourceID key) {
-    return getSource(key.getNamespace().orElse(ANY_NAMESPACE_MAP_KEY))
+    return getSource(key.getNamespace().orElse(ALL_NAMESPACES_MAP_KEY))
         .map(c -> c.remove(key))
         .orElse(null);
   }
 
   @Override
   public void put(ResourceID key, T resource) {
-    getSource(key.getNamespace().orElse(ANY_NAMESPACE_MAP_KEY))
+    getSource(key.getNamespace().orElse(ALL_NAMESPACES_MAP_KEY))
         .ifPresentOrElse(c -> c.put(key, resource),
             () -> log.warn(
                 "Cannot put resource in the cache. No related cache found: {}. Resource: {}",
@@ -147,6 +175,7 @@ public void put(ResourceID key, T resource) {
 
   @Override
   public void addIndexers(Map<String, Function<T, List<String>>> indexers) {
+    this.indexers.putAll(indexers);
     sources.values().forEach(s -> s.addIndexers(indexers));
   }
 
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/ManagedInformerEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/ManagedInformerEventSource.java
index 95aba48ad7..98e445032c 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/ManagedInformerEventSource.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/ManagedInformerEventSource.java
@@ -3,6 +3,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.stream.Stream;
@@ -15,6 +16,7 @@
 import io.fabric8.kubernetes.client.dsl.MixedOperation;
 import io.fabric8.kubernetes.client.dsl.Resource;
 import io.fabric8.kubernetes.client.informers.ResourceEventHandler;
+import io.javaoperatorsdk.operator.api.config.NamespaceChangeable;
 import io.javaoperatorsdk.operator.api.config.ResourceConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.RecentOperationCacheFiller;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
@@ -24,7 +26,8 @@
 
 public abstract class ManagedInformerEventSource<R extends HasMetadata, P extends HasMetadata, C extends ResourceConfiguration<R>>
     extends CachingEventSource<R, P>
-    implements ResourceEventHandler<R>, IndexerResourceCache<R>, RecentOperationCacheFiller<R> {
+    implements ResourceEventHandler<R>, IndexerResourceCache<R>, RecentOperationCacheFiller<R>,
+    NamespaceChangeable {
 
   private static final Logger log = LoggerFactory.getLogger(ManagedInformerEventSource.class);
 
@@ -60,6 +63,13 @@ protected InformerManager<R, C> manager() {
     return (InformerManager<R, C>) cache;
   }
 
+  @Override
+  public void changeNamespaces(Set<String> namespaces) {
+    if (allowsNamespaceChanges()) {
+      manager().changeNamespaces(namespaces);
+    }
+  }
+
   @Override
   public void start() {
     manager().start();
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/MockKubernetesClient.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/MockKubernetesClient.java
index 0c2db0f6b1..b282e11fe5 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/MockKubernetesClient.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/MockKubernetesClient.java
@@ -11,6 +11,7 @@
 import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation;
 import io.fabric8.kubernetes.client.dsl.Resource;
 import io.fabric8.kubernetes.client.informers.SharedIndexInformer;
+import io.fabric8.kubernetes.client.informers.cache.Indexer;
 
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyLong;
@@ -36,6 +37,8 @@ public static <T extends HasMetadata> KubernetesClient client(Class<T> clazz) {
     when(resources.inAnyNamespace()).thenReturn(inAnyNamespace);
     when(inAnyNamespace.withLabelSelector(nullable(String.class))).thenReturn(filterable);
     SharedIndexInformer<T> informer = mock(SharedIndexInformer.class);
+    Indexer mockIndexer = mock(Indexer.class);
+    when(informer.getIndexer()).thenReturn(mockIndexer);
     when(filterable.runnableInformer(anyLong())).thenReturn(informer);
     when(client.resources(clazz)).thenReturn(resources);
 
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventSourceManagerTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventSourceManagerTest.java
index f09605d095..ecf9fae52d 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventSourceManagerTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventSourceManagerTest.java
@@ -8,11 +8,13 @@
 import io.javaoperatorsdk.operator.MockKubernetesClient;
 import io.javaoperatorsdk.operator.OperatorException;
 import io.javaoperatorsdk.operator.api.config.MockControllerConfiguration;
+import io.javaoperatorsdk.operator.api.config.informer.InformerConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
 import io.javaoperatorsdk.operator.processing.Controller;
 import io.javaoperatorsdk.operator.processing.event.source.CachingEventSource;
 import io.javaoperatorsdk.operator.processing.event.source.EventSource;
 import io.javaoperatorsdk.operator.processing.event.source.controller.ControllerResourceEventSource;
+import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
 import io.javaoperatorsdk.operator.processing.event.source.timer.TimerEventSource;
 import io.javaoperatorsdk.operator.sample.simple.TestCustomResource;
 
@@ -21,17 +23,12 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.Mockito.eq;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.*;
 
 @SuppressWarnings({"rawtypes", "unchecked"})
 class EventSourceManagerTest {
 
-  private final EventProcessor eventHandler = mock(EventProcessor.class);
-  private final EventSourceManager eventSourceManager = new EventSourceManager(eventHandler);
+  private final EventSourceManager eventSourceManager = initManager();
 
   @Test
   public void registersEventSource() {
@@ -42,7 +39,7 @@ public void registersEventSource() {
     Set<EventSource> registeredSources = eventSourceManager.getRegisteredEventSources();
     assertThat(registeredSources).contains(eventSource);
 
-    verify(eventSource, times(1)).setEventHandler(eq(eventSourceManager.getEventHandler()));
+    verify(eventSource, times(1)).setEventHandler(eventSourceManager.getEventHandler());
   }
 
   @Test
@@ -139,6 +136,34 @@ void retrievingAnEventSourceWhenMultipleAreRegisteredForATypeShouldRequireAQuali
         eventSource);
   }
 
+  @Test
+  void changesNamespacesOnControllerAndInformerEventSources() {
+    String newNamespaces = "new-namespace";
+
+    final var configuration = MockControllerConfiguration.forResource(HasMetadata.class);
+    final Controller controller = new Controller(mock(Reconciler.class), configuration,
+        MockKubernetesClient.client(HasMetadata.class));
+
+    EventSources eventSources = spy(new EventSources());
+    var controllerResourceEventSourceMock = mock(ControllerResourceEventSource.class);
+    doReturn(controllerResourceEventSourceMock).when(eventSources).controllerResourceEventSource();
+    when(controllerResourceEventSourceMock.allowsNamespaceChanges()).thenCallRealMethod();
+    var manager = new EventSourceManager(controller, eventSources);
+
+    InformerConfiguration informerConfigurationMock = mock(InformerConfiguration.class);
+    when(informerConfigurationMock.followControllerNamespaceChanges()).thenReturn(true);
+    InformerEventSource informerEventSource = mock(InformerEventSource.class);
+    when(informerEventSource.resourceType()).thenReturn(TestCustomResource.class);
+    when(informerEventSource.getConfiguration()).thenReturn(informerConfigurationMock);
+    when(informerEventSource.allowsNamespaceChanges()).thenCallRealMethod();
+    manager.registerEventSource("ies", informerEventSource);
+
+    manager.changeNamespaces(Set.of(newNamespaces));
+
+    verify(informerEventSource, times(1)).changeNamespaces(Set.of(newNamespaces));
+    verify(controllerResourceEventSourceMock, times(1)).changeNamespaces(Set.of(newNamespaces));
+  }
+
   private EventSourceManager initManager() {
     final var configuration = MockControllerConfiguration.forResource(HasMetadata.class);
     final Controller controller = new Controller(mock(Reconciler.class), configuration,
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventSourcesTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventSourcesTest.java
index bc5e4551c7..3bc223653f 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventSourcesTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventSourcesTest.java
@@ -1,12 +1,21 @@
 package io.javaoperatorsdk.operator.processing.event;
 
+import java.util.Set;
+import java.util.stream.Collectors;
+
 import org.junit.jupiter.api.Test;
 
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.MockKubernetesClient;
+import io.javaoperatorsdk.operator.api.config.MockControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.processing.Controller;
 import io.javaoperatorsdk.operator.processing.event.source.EventSource;
 
 import static org.junit.jupiter.api.Assertions.*;
 import static org.mockito.Mockito.mock;
 
+@SuppressWarnings({"unchecked", "rawtypes"})
 class EventSourcesTest {
 
   EventSources eventSources = new EventSources();
@@ -19,4 +28,23 @@ void cannotAddTwoEventSourcesWithSameName() {
     });
   }
 
+  @Test
+  void allEventSourcesShouldReturnAll() {
+    // initial state doesn't have ControllerResourceEventSource
+    assertEquals(Set.of(eventSources.retryEventSource()), eventSources.allEventSources().collect(
+        Collectors.toSet()));
+    final var configuration = MockControllerConfiguration.forResource(HasMetadata.class);
+    final var controller = new Controller(mock(Reconciler.class), configuration,
+        MockKubernetesClient.client(HasMetadata.class));
+    eventSources.initControllerEventSource(controller);
+    assertEquals(
+        Set.of(eventSources.retryEventSource(), eventSources.controllerResourceEventSource()),
+        eventSources.allEventSources().collect(Collectors.toSet()));
+    final var source = mock(EventSource.class);
+    eventSources.add("foo", source);
+    assertEquals(Set.of(eventSources.retryEventSource(),
+        eventSources.controllerResourceEventSource(), source),
+        eventSources.allEventSources().collect(Collectors.toSet()));
+  }
+
 }
diff --git a/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/OperatorExtension.java b/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/OperatorExtension.java
index d95e9466a7..5f72a840ef 100644
--- a/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/OperatorExtension.java
+++ b/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/OperatorExtension.java
@@ -3,7 +3,9 @@
 import java.io.InputStream;
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -15,6 +17,7 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.LocalPortForward;
 import io.javaoperatorsdk.operator.Operator;
+import io.javaoperatorsdk.operator.RegisteredController;
 import io.javaoperatorsdk.operator.api.config.ConfigurationService;
 import io.javaoperatorsdk.operator.api.config.ControllerConfigurationOverrider;
 import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
@@ -32,6 +35,7 @@ public class OperatorExtension extends AbstractOperatorExtension {
   private final List<ReconcilerSpec> reconcilers;
   private List<PortFowardSpec> portForwards;
   private List<LocalPortForward> localPortForwards;
+  private Map<Reconciler, RegisteredController> registeredControllers;
 
   private OperatorExtension(
       ConfigurationService configurationService,
@@ -53,6 +57,7 @@ private OperatorExtension(
     this.portForwards = portForwards;
     this.localPortForwards = new ArrayList<>(portForwards.size());
     this.operator = new Operator(getKubernetesClient(), this.configurationService);
+    this.registeredControllers = new HashMap<>();
   }
 
   /**
@@ -85,6 +90,11 @@ public <T extends Reconciler> T getReconcilerOfType(Class<T> type) {
             () -> new IllegalArgumentException("Unable to find a reconciler of type: " + type));
   }
 
+  public RegisteredController getRegisteredControllerForReconcile(
+      Class<? extends Reconciler> type) {
+    return registeredControllers.get(getReconcilerOfType(type));
+  }
+
   @SuppressWarnings("unchecked")
   protected void before(ExtensionContext context) {
     super.before(context);
@@ -133,7 +143,8 @@ protected void before(ExtensionContext context) {
         ((KubernetesClientAware) ref.reconciler).setKubernetesClient(kubernetesClient);
       }
 
-      this.operator.register(ref.reconciler, oconfig.build());
+      var registeredController = this.operator.register(ref.reconciler, oconfig.build());
+      registeredControllers.put(ref.reconciler, registeredController);
     }
 
     LOGGER.debug("Starting the operator locally");
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/ChangeNamespaceIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/ChangeNamespaceIT.java
new file mode 100644
index 0000000000..56802ca8a3
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/ChangeNamespaceIT.java
@@ -0,0 +1,112 @@
+package io.javaoperatorsdk.operator;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.Set;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.api.model.Namespace;
+import io.fabric8.kubernetes.api.model.NamespaceBuilder;
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.javaoperatorsdk.operator.junit.OperatorExtension;
+import io.javaoperatorsdk.operator.sample.changenamespace.ChangeNamespaceTestCustomResource;
+import io.javaoperatorsdk.operator.sample.changenamespace.ChangeNamespaceTestReconciler;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+class ChangeNamespaceIT {
+
+  public static final String TEST_RESOURCE_NAME_1 = "test1";
+  public static final String TEST_RESOURCE_NAME_2 = "test2";
+  public static final String TEST_RESOURCE_NAME_3 = "test3";
+  public static final String ADDITIONAL_TEST_NAMESPACE = "additional-test-namespace";
+  @RegisterExtension
+  OperatorExtension operator =
+      OperatorExtension.builder().withReconciler(new ChangeNamespaceTestReconciler()).build();
+
+  @Test
+  void addNewAndRemoveOldNamespaceTest() {
+    try {
+      operator.create(ChangeNamespaceTestCustomResource.class,
+          customResource(TEST_RESOURCE_NAME_1));
+
+      await().pollDelay(Duration.ofMillis(100)).untilAsserted(() -> assertThat(
+          operator.get(ChangeNamespaceTestCustomResource.class, TEST_RESOURCE_NAME_1)
+              .getStatus().getNumberOfStatusUpdates()).isEqualTo(2));
+
+      client().namespaces().create(additionalTestNamespace());
+      createResourceInTestNamespace();
+
+      await().pollDelay(Duration.ofMillis(200)).untilAsserted(
+          () -> assertThat(client().resources(ChangeNamespaceTestCustomResource.class)
+              .inNamespace(ADDITIONAL_TEST_NAMESPACE)
+              .withName(TEST_RESOURCE_NAME_2).get().getStatus().getNumberOfStatusUpdates())
+                  .isZero());
+
+      // adding additional namespace
+      RegisteredController registeredController =
+          operator.getRegisteredControllerForReconcile(ChangeNamespaceTestReconciler.class);
+      registeredController
+          .changeNamespaces(Set.of(operator.getNamespace(), ADDITIONAL_TEST_NAMESPACE));
+
+      await().untilAsserted(
+          () -> assertThat(client().resources(ChangeNamespaceTestCustomResource.class)
+              .inNamespace(ADDITIONAL_TEST_NAMESPACE)
+              .withName(TEST_RESOURCE_NAME_2).get().getStatus().getNumberOfStatusUpdates())
+                  .withFailMessage("Informers don't watch the new namespace")
+                  .isEqualTo(2));
+
+      // removing a namespace
+      registeredController.changeNamespaces(Set.of(ADDITIONAL_TEST_NAMESPACE));
+
+      operator.create(ChangeNamespaceTestCustomResource.class,
+          customResource(TEST_RESOURCE_NAME_3));
+      await().pollDelay(Duration.ofMillis(200))
+          .untilAsserted(() -> assertThat(
+              operator.get(ChangeNamespaceTestCustomResource.class, TEST_RESOURCE_NAME_3)
+                  .getStatus().getNumberOfStatusUpdates()).isZero());
+
+
+      ConfigMap firstMap = operator.get(ConfigMap.class, TEST_RESOURCE_NAME_1);
+      firstMap.setData(Map.of("data", "newdata"));
+      operator.replace(ConfigMap.class, firstMap);
+
+      assertThat(operator.get(ChangeNamespaceTestCustomResource.class, TEST_RESOURCE_NAME_1)
+          .getStatus().getNumberOfStatusUpdates())
+              .withFailMessage("ConfigMap change induced a reconcile")
+              .isEqualTo(2);
+
+    } finally {
+      client().namespaces().delete(additionalTestNamespace());
+    }
+  }
+
+  private void createResourceInTestNamespace() {
+    var res = customResource(TEST_RESOURCE_NAME_2);
+    client().resources(ChangeNamespaceTestCustomResource.class)
+        .inNamespace(ADDITIONAL_TEST_NAMESPACE)
+        .create(res);
+  }
+
+  private KubernetesClient client() {
+    return operator.getKubernetesClient();
+  }
+
+  private Namespace additionalTestNamespace() {
+    return new NamespaceBuilder().withMetadata(new ObjectMetaBuilder()
+        .withName(ADDITIONAL_TEST_NAMESPACE)
+        .build()).build();
+  }
+
+  private ChangeNamespaceTestCustomResource customResource(String name) {
+    ChangeNamespaceTestCustomResource customResource = new ChangeNamespaceTestCustomResource();
+    customResource.setMetadata(
+        new ObjectMetaBuilder().withName(name).build());
+    return customResource;
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/changenamespace/ChangeNamespaceTestCustomResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/changenamespace/ChangeNamespaceTestCustomResource.java
new file mode 100644
index 0000000000..2b3b2295cd
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/changenamespace/ChangeNamespaceTestCustomResource.java
@@ -0,0 +1,18 @@
+package io.javaoperatorsdk.operator.sample.changenamespace;
+
+import io.fabric8.kubernetes.api.model.Namespaced;
+import io.fabric8.kubernetes.client.CustomResource;
+import io.fabric8.kubernetes.model.annotation.Group;
+import io.fabric8.kubernetes.model.annotation.Version;
+
+@Group("sample.javaoperatorsdk")
+@Version("v1")
+public class ChangeNamespaceTestCustomResource
+    extends CustomResource<Void, ChangeNamespaceTestCustomResourceStatus>
+    implements Namespaced {
+
+  @Override
+  protected ChangeNamespaceTestCustomResourceStatus initStatus() {
+    return new ChangeNamespaceTestCustomResourceStatus();
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/changenamespace/ChangeNamespaceTestCustomResourceStatus.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/changenamespace/ChangeNamespaceTestCustomResourceStatus.java
new file mode 100644
index 0000000000..a3858cda05
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/changenamespace/ChangeNamespaceTestCustomResourceStatus.java
@@ -0,0 +1,16 @@
+package io.javaoperatorsdk.operator.sample.changenamespace;
+
+public class ChangeNamespaceTestCustomResourceStatus {
+
+  private int numberOfStatusUpdates = 0;
+
+  public int getNumberOfStatusUpdates() {
+    return numberOfStatusUpdates;
+  }
+
+  public ChangeNamespaceTestCustomResourceStatus setNumberOfStatusUpdates(
+      int numberOfStatusUpdates) {
+    this.numberOfStatusUpdates = numberOfStatusUpdates;
+    return this;
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/changenamespace/ChangeNamespaceTestReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/changenamespace/ChangeNamespaceTestReconciler.java
new file mode 100644
index 0000000000..eb8dff9e7f
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/changenamespace/ChangeNamespaceTestReconciler.java
@@ -0,0 +1,76 @@
+package io.javaoperatorsdk.operator.sample.changenamespace;
+
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.javaoperatorsdk.operator.api.config.informer.InformerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.*;
+import io.javaoperatorsdk.operator.junit.KubernetesClientAware;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
+import io.javaoperatorsdk.operator.support.TestExecutionInfoProvider;
+
+@ControllerConfiguration
+public class ChangeNamespaceTestReconciler
+    implements Reconciler<ChangeNamespaceTestCustomResource>, TestExecutionInfoProvider,
+    EventSourceInitializer<ChangeNamespaceTestCustomResource>, KubernetesClientAware {
+
+  private final AtomicInteger numberOfExecutions = new AtomicInteger(0);
+  private KubernetesClient client;
+
+  @Override
+  public Map<String, EventSource> prepareEventSources(
+      EventSourceContext<ChangeNamespaceTestCustomResource> context) {
+
+    InformerEventSource<ConfigMap, ChangeNamespaceTestCustomResource> configMapES =
+        new InformerEventSource<>(InformerConfiguration.from(ConfigMap.class, context)
+            .build(), context);
+
+    return EventSourceInitializer.nameEventSources(configMapES);
+  }
+
+  @Override
+  public UpdateControl<ChangeNamespaceTestCustomResource> reconcile(
+      ChangeNamespaceTestCustomResource primary,
+      Context<ChangeNamespaceTestCustomResource> context) {
+
+    var actualConfigMap = context.getSecondaryResource(ConfigMap.class);
+    if (actualConfigMap.isEmpty()) {
+      client.configMaps().inNamespace(primary.getMetadata().getNamespace())
+          .create(configMap(primary));
+    }
+
+    numberOfExecutions.addAndGet(1);
+
+    var statusUpdates = primary.getStatus().getNumberOfStatusUpdates();
+    primary.getStatus().setNumberOfStatusUpdates(statusUpdates + 1);
+    return UpdateControl.updateStatus(primary);
+  }
+
+  public int getNumberOfExecutions() {
+    return numberOfExecutions.get();
+  }
+
+  private ConfigMap configMap(ChangeNamespaceTestCustomResource primary) {
+    ConfigMap configMap = new ConfigMap();
+    configMap.setMetadata(new ObjectMetaBuilder().withName(primary.getMetadata().getName())
+        .withNamespace(primary.getMetadata().getNamespace())
+        .build());
+    configMap.setData(Map.of("data", primary.getMetadata().getName()));
+    configMap.addOwnerReference(primary);
+    return configMap;
+  }
+
+  @Override
+  public KubernetesClient getKubernetesClient() {
+    return client;
+  }
+
+  @Override
+  public void setKubernetesClient(KubernetesClient kubernetesClient) {
+    this.client = kubernetesClient;
+  }
+}
diff --git a/sample-operators/tomcat-operator/src/main/java/io/javaoperatorsdk/operator/sample/WebappReconciler.java b/sample-operators/tomcat-operator/src/main/java/io/javaoperatorsdk/operator/sample/WebappReconciler.java
index ffd530713c..49325def03 100644
--- a/sample-operators/tomcat-operator/src/main/java/io/javaoperatorsdk/operator/sample/WebappReconciler.java
+++ b/sample-operators/tomcat-operator/src/main/java/io/javaoperatorsdk/operator/sample/WebappReconciler.java
@@ -59,7 +59,7 @@ public Map<String, EventSource> prepareEventSources(EventSourceContext<Webapp> c
             .collect(Collectors.toSet());
 
     InformerConfiguration<Tomcat> configuration =
-        InformerConfiguration.from(Tomcat.class)
+        InformerConfiguration.from(Tomcat.class, context)
             .withSecondaryToPrimaryMapper(webappsMatchingTomcatName)
             .build();
     return EventSourceInitializer
diff --git a/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageReconciler.java b/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageReconciler.java
index 7ff94edc4b..453fd1671f 100644
--- a/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageReconciler.java
+++ b/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageReconciler.java
@@ -51,19 +51,19 @@ public WebPageReconciler(KubernetesClient kubernetesClient) {
   @Override
   public Map<String, EventSource> prepareEventSources(EventSourceContext<WebPage> context) {
     var configMapEventSource =
-        new InformerEventSource<>(InformerConfiguration.from(ConfigMap.class)
+        new InformerEventSource<>(InformerConfiguration.from(ConfigMap.class, context)
             .withLabelSelector(LOW_LEVEL_LABEL_KEY)
             .build(), context);
     var deploymentEventSource =
-        new InformerEventSource<>(InformerConfiguration.from(Deployment.class)
+        new InformerEventSource<>(InformerConfiguration.from(Deployment.class, context)
             .withLabelSelector(LOW_LEVEL_LABEL_KEY)
             .build(), context);
     var serviceEventSource =
-        new InformerEventSource<>(InformerConfiguration.from(Service.class)
+        new InformerEventSource<>(InformerConfiguration.from(Service.class, context)
             .withLabelSelector(LOW_LEVEL_LABEL_KEY)
             .build(), context);
     var ingressEventSource =
-        new InformerEventSource<>(InformerConfiguration.from(Ingress.class)
+        new InformerEventSource<>(InformerConfiguration.from(Ingress.class, context)
             .withLabelSelector(LOW_LEVEL_LABEL_KEY)
             .build(), context);
     return EventSourceInitializer.nameEventSources(configMapEventSource, deploymentEventSource,