apache · ctrlaltdilj · Apr 6, 2025
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -121,15 +121,24 @@ jobs:
           - test_multi_sessionjob.sh
           - test_autoscaler.sh
           - test_dynamic_config.sh
+          - test_dynamic_flink_conf.sh
         exclude:
           - mode: standalone
             test: test_autoscaler.sh
           - mode: standalone
             test: test_dynamic_config.sh
+          - mode: standalone
+            test: test_dynamic_flink_conf.sh
           - flink-version: v1_16
             test: test_autoscaler.sh
           - flink-version: v1_16
             test: test_dynamic_config.sh
+          - flink-version: v1_17
+            test: test_dynamic_config.sh
+          - flink-version: v1_18
+            test: test_dynamic_config.sh
+          - flink-version: v1_16
+            test: test_dynamic_flink_conf.sh
 
     uses: ./.github/workflows/e2e.yaml
     with:
@@ -156,6 +165,7 @@ jobs:
           - test_application_kubernetes_ha.sh
           - test_application_operations.sh
           - test_dynamic_config.sh
+          - test_dynamic_flink_conf.sh
           - test_sessionjob_kubernetes_ha.sh
           - test_sessionjob_operations.sh
           - test_autoscaler.sh
@@ -169,6 +179,8 @@ jobs:
             test: test_autoscaler.sh
           - flink-version: v1_16
             test: test_dynamic_config.sh
+          - flink-version: v1_16
+            test: test_dynamic_flink_conf.sh
           - flink-version: v1_16
             test: test_flink_operator_ha.sh
           - flink-version: v1_16
@@ -177,6 +189,8 @@ jobs:
             test: test_batch_job.sh
           - flink-version: v1_17
             test: test_dynamic_config.sh
+          - flink-version: v1_17
+            test: test_dynamic_flink_conf.sh
           - flink-version: v1_17
             test: test_flink_operator_ha.sh
           - flink-version: v1_17
@@ -185,6 +199,8 @@ jobs:
             test: test_batch_job.sh
           - flink-version: v1_18
             test: test_dynamic_config.sh
+          - flink-version: v1_18
+            test: test_dynamic_flink_conf.sh
           - flink-version: v1_18
             test: test_flink_operator_ha.sh
           - flink-version: v1_18

diff --git a/.github/workflows/e2e.yaml b/.github/workflows/e2e.yaml
@@ -75,6 +75,9 @@ jobs:
            sed -i "s/# kubernetes.operator.leader-election.lease-name: flink-operator-lease/kubernetes.operator.leader-election.lease-name: flink-operator-lease/" helm/flink-kubernetes-operator/conf/flink-conf.yaml
            sed -i "s/replicas: 1/replicas: 2/" helm/flink-kubernetes-operator/values.yaml
           fi
+          if [[ "${{ inputs.test }}" == "test_dynamic_config.sh" ]]; then          
+           sed -i "s/flink-conf.yaml: |+/config.yaml: |+/" helm/flink-kubernetes-operator/values.yaml
+          fi
           helm --debug install flink-kubernetes-operator -n ${{ inputs.namespace }} helm/flink-kubernetes-operator --set image.repository=flink-kubernetes-operator --set image.tag=ci-latest ${{ steps.namespace-creator.outputs.EXTRA_HELM_INSTALL_ARGS }} 
           kubectl wait --for=condition=Available --timeout=120s -n ${{ inputs.namespace }} deploy/flink-kubernetes-operator
           kubectl get pods -n ${{ inputs.namespace }}

diff --git a/docs/content/docs/operations/helm.md b/docs/content/docs/operations/helm.md
@@ -112,6 +112,8 @@ The configurable parameters of the Helm chart and which default values as detail
 | defaultConfiguration.create                    | Whether to enable default configuration to create for flink-kubernetes-operator.                                                                               | true                                                                                                                                                                                                                                                                                           |
 | defaultConfiguration.append                    | Whether to append configuration files with configs.                                                                                                            | true                                                                                                                                                                                                                                                                                           |
 | defaultConfiguration.flink-conf.yaml           | The default configuration of flink-conf.yaml.                                                                                                                  | kubernetes.operator.metrics.reporter.slf4j.factory.class: org.apache.flink.metrics.slf4j.Slf4jReporterFactory<br/>kubernetes.operator.metrics.reporter.slf4j.interval: 5 MINUTE<br/>kubernetes.operator.reconcile.interval: 15 s<br/>kubernetes.operator.observer.progress-check.interval: 5 s |
+| defaultConfiguration.config.yaml           | The newer configuration file format for flink that will enforced in Flink 2.0. Note this was introudced in flink 1.19.                                                                                                                   | kubernetes.operator.metrics.reporter.slf4j.factory.class: org.apache.flink.metrics.slf4j.Slf4jReporterFactory<br/>kubernetes.operator.metrics.reporter.slf4j.interval: 5 MINUTE<br/>kubernetes.operator.reconcile.interval: 15 s<br/>kubernetes.operator.observer.progress-check.interval: 5 s |
+
 | defaultConfiguration.log4j-operator.properties | The default configuration of log4j-operator.properties.                                                                                                        |                                                                                                                                                                                                                                                                                                |
 | defaultConfiguration.log4j-console.properties  | The default configuration of log4j-console.properties.                                                                                                         |                                                                                                                                                                                                                                                                                                |
 | metrics.port                                   | The metrics port on the container for default configuration.                                                                                                   |                                                                                                                                                                                                                                                                                                |

diff --git a/e2e-tests/test_dynamic_config.sh b/e2e-tests/test_dynamic_config.sh
@@ -34,7 +34,7 @@ echo "Current operator pod is ${operator_pod}"
 create_namespace dynamic
 
 kubectl config set-context --current --namespace="${operator_namespace}"
-patch_flink_config '{"data": {"flink-conf.yaml": "kubernetes.operator.watched.namespaces: default,flink,dynamic"}}'
+patch_flink_config '{"data": {"config.yaml": "kubernetes.operator.watched.namespaces: default,flink,dynamic"}}'
 wait_for_operator_logs "${operator_pod}" "Setting default configuration to {kubernetes.operator.watched.namespaces=default,flink,dynamic}" ${TIMEOUT} || exit 1
 
 echo "Successfully run the dynamic property test"
diff --git a/e2e-tests/test_dynamic_flink_conf.sh b/e2e-tests/test_dynamic_flink_conf.sh
@@ -0,0 +1,40 @@
+#!/usr/bin/env bash
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+# This script tests the operator dynamic config on watched namespace:
+# 1. Create a new namespace
+# 2. Change the watched namespaces by patching on the flink-config-override
+# 3. Monitor the operator log to find the watched namespace changed info
+SCRIPT_DIR=$(dirname "$(readlink -f "$0")")
+source "${SCRIPT_DIR}/utils.sh"
+
+on_exit operator_cleanup_and_exit
+
+TIMEOUT=360
+
+operator_namespace=$(get_operator_pod_namespace)
+operator_pod=$(get_operator_pod_name)
+echo "Current operator pod is ${operator_pod}"
+create_namespace dynamic
+
+kubectl config set-context --current --namespace="${operator_namespace}"
+patch_flink_config '{"data": {"flink-conf.yaml": "kubernetes.operator.watched.namespaces: default,flink,dynamic"}}'
+wait_for_operator_logs "${operator_pod}" "Setting default configuration to {kubernetes.operator.watched.namespaces=default,flink,dynamic}" ${TIMEOUT} || exit 1
+
+echo "Successfully run the dynamic property test"
diff --git a/helm/flink-kubernetes-operator/templates/flink-operator.yaml b/helm/flink-kubernetes-operator/templates/flink-operator.yaml
@@ -235,8 +235,13 @@ spec:
           configMap:
             name: flink-operator-config
             items:
+            {{- if hasKey .Values.defaultConfiguration "flink-conf.yaml" }}
               - key: flink-conf.yaml
                 path: flink-conf.yaml
+            {{- else }}
+              - key: config.yaml
+                path: config.yaml
+            {{- end }}
               - key: log4j-operator.properties
                 path: log4j-operator.properties
               - key: log4j-console.properties
@@ -271,11 +276,25 @@ metadata:
   labels:
     {{- include "flink-operator.labels" . | nindent 4 }}
 data:
+  config.yaml: |+
+{{- if .Values.defaultConfiguration.append }}
+  {{- $.Files.Get "conf/flink-conf.yaml"  | nindent 4 -}}
+{{- end }}
+{{- if hasKey (.Values.defaultConfiguration) "config.yaml" }}
+  {{- index (.Values.defaultConfiguration) "config.yaml" | nindent 4 -}}
+{{- end }}
+{{- if .Values.watchNamespaces }}
+    kubernetes.operator.watched.namespaces: {{ join "," .Values.watchNamespaces  }}
+{{- end }}
+{{- if index .Values "operatorHealth" }}
+    kubernetes.operator.health.probe.enabled: true
+    kubernetes.operator.health.probe.port: {{ .Values.operatorHealth.port }}
+{{- end }}
   flink-conf.yaml: |+
 {{- if .Values.defaultConfiguration.append }}
   {{- $.Files.Get "conf/flink-conf.yaml"  | nindent 4 -}}
 {{- end }}
-{{- if index (.Values.defaultConfiguration) "flink-conf.yaml" }}
+{{- if hasKey (.Values.defaultConfiguration) "flink-conf.yaml" }}
   {{- index (.Values.defaultConfiguration) "flink-conf.yaml" | nindent 4 -}}
 {{- end }}
 {{- if .Values.watchNamespaces }}