Replace dynamic-config informer with mounted-ConfigMap file watcher

Rework the dynamic-config feature so the operator no longer runs a
dedicated Operator instance with a ConfigMap informer. The dynamic-config
ConfigMap is now mounted into the operator pod as a volume, and a new
DynamicConfigMonitor periodically re-reads the properties file from disk,
refreshing SparkOperatorConfManager and the watched-namespaces updater on
change.

- Add DynamicConfigMonitor with start/stop/isRunning lifecycle
- Simplify SparkOperator from List<Operator> to a single Operator plus an
  optional DynamicConfigMonitor; remove SparkOperatorConfigMapReconciler
- Replace dynamicConfig.selector with dynamicConfig.filePath and
  dynamicConfig.reloadIntervalSeconds
- Helm: drop configmaps RBAC Role/RoleBinding, mount the dynamic ConfigMap
  volume, render a single spark-operator-dynamic.properties key
- Probes: switch to a single Operator and include the monitor's running
  state in /healthz and /readyz

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: csviri

build-tools/helm/spark-kubernetes-operator/templates/operator-rbac.yaml

@@ -118,37 +118,6 @@ metadata:
 {{- template "spark-operator.operatorRbacRules" $ }}
 ---
 {{- end }}
-{{- if .Values.operatorConfiguration.dynamicConfig.enable }}
-apiVersion: rbac.authorization.k8s.io/v1
-kind: Role
-metadata:
-  name: {{ .Values.operatorRbac.configManagement.roleName }}
-  namespace: {{ .Release.Namespace }}
-{{- template "spark-operator.rbacLabelsAnnotations" $ }}
-rules:
-  - apiGroups:
-      - ""
-    resources:
-      - configmaps
-    verbs:
-      - '*'
----
-apiVersion: rbac.authorization.k8s.io/v1
-kind: RoleBinding
-metadata:
-  name: {{ .Values.operatorRbac.configManagement.roleBindingName }}
-  namespace: {{ .Release.Namespace }}
-{{- template "spark-operator.rbacLabelsAnnotations" $ }}
-roleRef:
-  apiGroup: rbac.authorization.k8s.io
-  kind: Role
-  name: {{ .Values.operatorRbac.configManagement.roleName }}
-subjects:
-  - kind: ServiceAccount
-    name: {{ .Values.operatorRbac.serviceAccount.name }}
-    namespace: {{ .Release.Namespace }}
----
-{{- end }}
 
 {{- $workloadResources := .Values.workloadResources -}}
 {{- $systemNs := .Release.Namespace -}}

build-tools/helm/spark-kubernetes-operator/templates/spark-operator.yaml

@@ -148,6 +148,11 @@ spec:
               mountPath: /opt/spark-operator/conf
             - name: logs-volume
               mountPath: /opt/spark-operator/logs
+            {{- if .Values.operatorConfiguration.dynamicConfig.enable }}
+            - name: spark-operator-dynamic-config-volume
+              mountPath: /opt/spark-operator/dynamic-conf
+              readOnly: true
+            {{- end }}
             {{- with .Values.operatorDeployment.operatorPod.operatorContainer.volumeMounts }}
               {{- toYaml . | nindent 12 }}
             {{- end }}
@@ -170,6 +175,11 @@ spec:
             name: spark-kubernetes-operator-configuration
         - name: logs-volume
           emptyDir: { }
+        {{- if .Values.operatorConfiguration.dynamicConfig.enable }}
+        - name: spark-operator-dynamic-config-volume
+          configMap:
+            name: spark-kubernetes-operator-dynamic-configuration
+        {{- end }}
         {{- with .Values.operatorDeployment.operatorPod.volumes }}
           {{- toYaml . | nindent 8 }}
         {{- end }}
@@ -215,8 +225,9 @@ metadata:
     {{- include "spark-operator.dynamicConfigLabels" . | nindent 4 }}
   annotations:
     {{- toYaml .Values.operatorConfiguration.dynamicConfig.annotations | nindent 4 }}
-{{- with .Values.operatorConfiguration.dynamicConfig.data }}
 data:
-  {{- toYaml . | nindent 2 }}
-{{- end }}
+  spark-operator-dynamic.properties: |
+    {{- range $key, $value := .Values.operatorConfiguration.dynamicConfig.data }}
+    {{ $key }}={{ $value }}
+    {{- end }}
 {{- end }}

docs/config_properties.md

@@ -10,9 +10,10 @@
  | spark.kubernetes.operator.api.secondaryResourceCreateMaxAttempts | Long | 3 | false | Maximal number of retry attempts of requesting secondary resource for Spark application. This would be performed on top of k8s client spark.kubernetes.operator.retry.maxAttempts to overcome potential conflicting reconcile on the same SparkApplication, as well as API server errors (408/500/502/503/504) and network-level timeouts. Exponential backoff with jitter is applied before retrying on 409 (Conflict) and 429 (Too Many Requests) responses. Other retryable errors are retried immediately. This should be positive number. | 
  | spark.kubernetes.operator.api.secondaryResourceCreateMaxBackoffMillis | Long | 40000 | false | Maximum backoff (in milliseconds) between retries when creating secondary resources for Spark application. | 
  | spark.kubernetes.operator.api.statusPatchMaxAttempts | Long | 3 | false | Maximal number of retry attempts of requests to k8s server for resource status update. This would be performed on top of k8s client spark.kubernetes.operator.retry.maxAttempts to overcome potential conflicting update on the same SparkApplication. This should be positive number. | 
- | spark.kubernetes.operator.dynamicConfig.enabled | Boolean | false | false | When enabled, operator would use config map as source of truth for config property override. The config map need to be created in spark.kubernetes.operator.namespace, and labeled with operator name. | 
+ | spark.kubernetes.operator.dynamicConfig.enabled | Boolean | false | false | When enabled, operator periodically reloads property overrides from a properties file mounted from a ConfigMap. The file path is controlled by spark.kubernetes.operator.dynamicConfig.filePath. | 
+ | spark.kubernetes.operator.dynamicConfig.filePath | String | /opt/spark-operator/dynamic-conf/spark-operator-dynamic.properties | false | Path of the properties file holding dynamic configuration overrides. Typically populated by mounting a ConfigMap as a volume. | 
  | spark.kubernetes.operator.dynamicConfig.reconcilerParallelism | Integer | 1 | false | Parallelism for dynamic config reconciler. Unbounded pool would be used if set to non-positive number. | 
- | spark.kubernetes.operator.dynamicConfig.selector | String | app.kubernetes.io/name=spark-kubernetes-operator,app.kubernetes.io/component=operator-dynamic-config-overrides | false | The selector str applied to dynamic config map. | 
+ | spark.kubernetes.operator.dynamicConfig.reloadIntervalSeconds | Long | 60 | false | Interval (in seconds) at which the dynamic config file is re-read. | 
  | spark.kubernetes.operator.health.probePort | Integer | 19091 | false | The port used for health/readiness check probe status. | 
  | spark.kubernetes.operator.health.sentinelExecutorPoolSize | Integer | 3 | false | Size of executor service in Sentinel Managers to check the health of sentinel resources. | 
  | spark.kubernetes.operator.health.sentinelResourceReconciliationDelaySeconds | Integer | 60 | true | Allowed max time(seconds) between spec update and reconciliation for sentinel resources. | 

gradle/libs.versions.toml

@@ -29,6 +29,7 @@ slf4j = "2.0.18"
 junit = "6.1.0"
 jacoco = "0.8.14"
 mockito = "5.23.0"
+awaitility = "4.3.0"
 
 # Build Analysis
 checkstyle = "13.4.2"
@@ -64,6 +65,7 @@ metrics-jvm = { group = "io.dropwizard.metrics", name = "metrics-jvm", version.r
 spark-core = { group = "org.apache.spark", name = "spark-core_2.13", version.ref = "spark"}
 spark-kubernetes = { group = "org.apache.spark", name = "spark-kubernetes_2.13", version.ref = "spark"}
 mockito-core = { group = "org.mockito", name = "mockito-core", version.ref = "mockito"}
+awaitility = { group = "org.awaitility", name = "awaitility", version.ref = "awaitility"}
 junit-bom = { group = "org.junit", name = "junit-bom", version.ref = "junit"}
 junit-jupiter = { group = "org.junit.jupiter", name = "junit-jupiter", version.ref = "junit"}
 junit-platform-launcher = { group = "org.junit.platform", name = "junit-platform-launcher"}

spark-operator/build.gradle

@@ -71,6 +71,7 @@ dependencies {
   testCompileOnly(libs.spotbugs.annotations)
   testImplementation(libs.mockito.core)
   testImplementation(libs.kube.api.test.client.inject)
+  testImplementation(libs.awaitility)
 }
 
 jar.dependsOn shadowJar

spark-operator/src/main/java/org/apache/spark/k8s/operator/SparkOperator.java

@@ -23,11 +23,13 @@
 import static org.apache.spark.k8s.operator.utils.Utils.getClusterStatusListener;
 import static org.apache.spark.k8s.operator.utils.Utils.getWatchedNamespaces;
 
+import java.nio.file.Paths;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -43,9 +45,9 @@
 import lombok.extern.slf4j.Slf4j;
 
 import org.apache.spark.k8s.operator.client.KubernetesClientFactory;
+import org.apache.spark.k8s.operator.config.DynamicConfigMonitor;
 import org.apache.spark.k8s.operator.config.SparkOperatorConf;
 import org.apache.spark.k8s.operator.config.SparkOperatorConfManager;
-import org.apache.spark.k8s.operator.config.SparkOperatorConfigMapReconciler;
 import org.apache.spark.k8s.operator.metrics.MetricsService;
 import org.apache.spark.k8s.operator.metrics.MetricsSystem;
 import org.apache.spark.k8s.operator.metrics.MetricsSystemFactory;
@@ -59,6 +61,7 @@
 import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
 import org.apache.spark.k8s.operator.utils.SparkClusterStatusRecorder;
 import org.apache.spark.k8s.operator.utils.StringUtils;
+import org.apache.spark.k8s.operator.utils.Utils;
 
 /**
  * Entry point for Spark Operator. Bootstrap the operator app by starting watch and reconciler for
@@ -67,7 +70,7 @@
  */
 @Slf4j
 public class SparkOperator {
-  private final List<Operator> registeredOperators;
+  private final Operator operator;
   private final KubernetesClient client;
   private final SparkAppSubmissionWorker appSubmissionWorker;
   private final SparkClusterSubmissionWorker clusterSubmissionWorker;
@@ -82,6 +85,7 @@ public class SparkOperator {
   private final MetricsService metricsService;
   private final ExecutorService metricsResourcesSingleThreadPool;
   private final ScheduledExecutorService periodicGcScheduler;
+  private final DynamicConfigMonitor dynamicConfigMonitor;
 
   /** Constructs a new SparkOperator, initializing all its components. */
   public SparkOperator() {
@@ -99,21 +103,19 @@ public SparkOperator() {
     this.watchedNamespaces = getWatchedNamespaces();
     this.sparkApplicationSentinelManager = new SentinelManager<>();
     this.sparkClusterSentinelManager = new SentinelManager<>();
-    this.registeredOperators = new ArrayList<>();
-    this.registeredOperators.add(registerSparkOperator());
+    this.operator = registerSparkOperator();
     if (SparkOperatorConf.LOG_CONF.getValue()) {
       for (var entry : SparkOperatorConfManager.INSTANCE.getAll().entrySet()) {
         log.info("{} = {}", entry.getKey(), entry.getValue());
       }
     }
-    if (SparkOperatorConf.DYNAMIC_CONFIG_ENABLED.getValue()) {
-      this.registeredOperators.add(registerSparkOperatorConfMonitor());
-    }
+    this.dynamicConfigMonitor = registerConfMonitor().orElse(null);
     this.metricsResourcesSingleThreadPool = Executors.newSingleThreadExecutor();
     this.probeService =
         new ProbeService(
-            registeredOperators,
+            operator,
             Arrays.asList(sparkApplicationSentinelManager, sparkClusterSentinelManager),
+            dynamicConfigMonitor,
             null);
     this.metricsService = new MetricsService(metricsSystem, metricsResourcesSingleThreadPool);
     long periodicGcIntervalSeconds = SparkOperatorConf.PERIODIC_GC_INTERVAL_SECONDS.getValue();
@@ -175,27 +177,20 @@ protected Operator registerSparkOperator() {
   }
 
   /**
-   * Registers a monitor for dynamic configuration changes via ConfigMaps.
+   * Registers a monitor for dynamic configuration changes via a mounted ConfigMap.
    *
-   * @return The Operator instance for the config monitor.
+   * @return The DynamicConfigMonitor wrapped in an Optional, empty if dynamic config is disabled.
    */
-  protected Operator registerSparkOperatorConfMonitor() {
-    Operator op = new Operator(this::overrideConfigMonitorConfigs);
-    String operatorNamespace = SparkOperatorConf.OPERATOR_NAMESPACE.getValue();
-    String confSelector = SparkOperatorConf.DYNAMIC_CONFIG_SELECTOR.getValue();
-    log.info(
-        "Starting conf monitor in namespace: {}, with selector: {}",
-        operatorNamespace,
-        confSelector);
-    op.register(
-        new SparkOperatorConfigMapReconciler(
-            this::updateWatchingNamespaces, unused -> getWatchedNamespaces()),
-        c -> {
-          c.withRateLimiter(SparkOperatorConf.getOperatorRateLimiter());
-          c.settingNamespaces(operatorNamespace);
-          c.withLabelSelector(confSelector);
-        });
-    return op;
+  protected Optional<DynamicConfigMonitor> registerConfMonitor() {
+    if (Boolean.FALSE.equals(SparkOperatorConf.DYNAMIC_CONFIG_ENABLED.getValue())) {
+      return Optional.empty();
+    }
+    return Optional.of(
+        new DynamicConfigMonitor(
+            Paths.get(SparkOperatorConf.DYNAMIC_CONFIG_FILE_PATH.getValue()),
+            Duration.ofSeconds(SparkOperatorConf.DYNAMIC_CONFIG_RELOAD_INTERVAL_SECONDS.getValue()),
+            Utils::getWatchedNamespaces,
+            this::updateWatchingNamespaces));
   }
 
   /**
@@ -263,23 +258,6 @@ protected void overrideOperatorConfigs(ConfigurationServiceOverrider overrider)
     overrider.withUseSSAToPatchPrimaryResource(false);
   }
 
-  /**
-   * Overrides the configuration for the dynamic config monitor.
-   *
-   * @param overrider The ConfigurationServiceOverrider to apply changes to.
-   */
-  protected void overrideConfigMonitorConfigs(ConfigurationServiceOverrider overrider) {
-    overrider.withKubernetesClient(client);
-    overrider.withConcurrentReconciliationThreads(
-        SparkOperatorConf.DYNAMIC_CONFIG_RECONCILER_PARALLELISM.getValue());
-    overrider.withStopOnInformerErrorDuringStartup(true);
-    overrider.withCloseClientOnStop(false);
-    overrider.withInformerStoppedHandler(
-        (informer, ex) ->
-            log.error("Dynamic config informer stopped: operator will not accept config updates."));
-    overrider.withUseSSAToPatchPrimaryResource(false);
-  }
-
   /**
    * Overrides the default configuration for individual controllers.
    *
@@ -328,8 +306,10 @@ public static void main(String[] args) {
     log.info("Java Version: {}", Runtime.version().toString());
     log.info("Built-in Spark Version: {}", org.apache.spark.package$.MODULE$.SPARK_VERSION());
     SparkOperator sparkOperator = new SparkOperator();
-    for (Operator operator : sparkOperator.registeredOperators) {
-      operator.start();
+    sparkOperator.operator.start();
+
+    if (sparkOperator.dynamicConfigMonitor != null) {
+      sparkOperator.dynamicConfigMonitor.start();
     }
     sparkOperator.probeService.start();
     // Single thread queue to ensure MetricsService starts after the MetricsSystem