protected boolean updateWatchingNamespaces(Set<String> namespaces) {
    if (watchedNamespaces.equals(namespaces)) {
      log.info("No watched namespace change detected");
      return false;
    }
    if (watchedNamespaces.isEmpty()) {
      log.info("Cannot update watch namespaces for operator started at cluster level.");
      return false;
    }
    if (namespaces == null || namespaces.isEmpty()) {
      log.error("Cannot updating namespaces to empty");
      return false;
    }
    registeredSparkControllers.forEach(
        c -> {
          if (c.allowsNamespaceChanges()) {
            log.info("Updating operator namespaces to {}", namespaces);
            c.changeNamespaces(namespaces);
          } else {
            log.error("Controller does not allow namespace change, skipping namespace change.");
          }
        });
    this.watchedNamespaces = new HashSet<>(namespaces);
    return true;
  }
@Test
  void testUpdateWatchedNamespacesWithDynamicConfigEnabled() {
    MetricsSystem mockMetricsSystem = mock(MetricsSystem.class);
    KubernetesClient mockClient = mock(KubernetesClient.class);
    var registeredController = mock(RegisteredController.class);
    when(registeredController.allowsNamespaceChanges()).thenReturn(true);
    boolean dynamicConfigEnabled = SparkOperatorConf.DYNAMIC_CONFIG_ENABLED.getValue();

    try (MockedStatic<MetricsSystemFactory> mockMetricsSystemFactory =
            mockStatic(MetricsSystemFactory.class);
        MockedStatic<KubernetesClientFactory> mockKubernetesClientFactory =
            mockStatic(KubernetesClientFactory.class);
        MockedStatic<Utils> mockUtils = mockStatic(Utils.class);
        MockedConstruction<Operator> operatorConstruction =
            mockConstruction(
                Operator.class,
                (mock, context) -> {
                  when(mock.register(any(SparkAppReconciler.class), any(Consumer.class)))
                      .thenReturn(registeredController);
                  when(mock.register(any(SparkClusterReconciler.class), any(Consumer.class)))
                      .thenReturn(registeredController);
                });
        MockedConstruction<SparkAppReconciler> sparkAppReconcilerConstruction =
            mockConstruction(SparkAppReconciler.class);
        MockedConstruction<SparkOperatorConfigMapReconciler> configReconcilerConstruction =
            mockConstruction(SparkOperatorConfigMapReconciler.class);
        MockedConstruction<ProbeService> probeServiceConstruction =
            mockConstruction(ProbeService.class);
        MockedConstruction<MetricsService> metricsServiceConstruction =
            mockConstruction(MetricsService.class);
        MockedConstruction<KubernetesMetricsInterceptor> interceptorMockedConstruction =
            mockConstruction(KubernetesMetricsInterceptor.class)) {
      setConfigKey(SparkOperatorConf.DYNAMIC_CONFIG_ENABLED, true);
      mockMetricsSystemFactory
          .when(MetricsSystemFactory::createMetricsSystem)
          .thenReturn(mockMetricsSystem);
      mockKubernetesClientFactory
          .when(() -> KubernetesClientFactory.buildKubernetesClient(any()))
          .thenReturn(mockClient);
      mockUtils.when(Utils::getWatchedNamespaces).thenReturn(Collections.singleton("namespace-1"));
      SparkOperator sparkOperator = new SparkOperator();
      Set<String> updatedNamespaces = Set.of("namespace-1", "namespace-2");
      Assertions.assertTrue(sparkOperator.updateWatchingNamespaces(updatedNamespaces));
      Assertions.assertEquals(updatedNamespaces, sparkOperator.watchedNamespaces);
      verify(registeredController).allowsNamespaceChanges();
      verify(registeredController).changeNamespaces(updatedNamespaces);
      verifyNoMoreInteractions(registeredController);
    } finally {
      setConfigKey(SparkOperatorConf.DYNAMIC_CONFIG_ENABLED, dynamicConfigEnabled);
    }
  }