Fixing configOption default value management and log message formatting

schongloo · schongloo · commit ebab27944a22 · 2025-07-14T22:11:58.000-07:00
diff --git a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/FlinkBlueGreenDeploymentConfigOptions.java b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/FlinkBlueGreenDeploymentConfigOptions.java
@@ -20,35 +20,39 @@
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.ConfigOptions;
 
+import java.time.Duration;
+
 /** Configuration options to be used by the Flink Blue/Green Deployments. */
 public class FlinkBlueGreenDeploymentConfigOptions {
 
-    public static final String BLUE_GREEN_CONF_PREFIX = "bluegreen.";
+    public static final String K8S_OP_CONF_PREFIX = "kubernetes.operator.";
+
+    public static final String BLUE_GREEN_CONF_PREFIX = K8S_OP_CONF_PREFIX + "bluegreen.";
 
     public static final int MIN_ABORT_GRACE_PERIOD_MS = 120000; // 2 mins
 
     public static ConfigOptions.OptionBuilder operatorConfig(String key) {
         return ConfigOptions.key(BLUE_GREEN_CONF_PREFIX + key);
     }
 
-    public static final ConfigOption<Integer> ABORT_GRACE_PERIOD_MS =
-            operatorConfig("abortGracePeriodMs")
-                    .intType()
-                    .defaultValue(0)
+    public static final ConfigOption<Duration> ABORT_GRACE_PERIOD =
+            operatorConfig("abort.grace-period")
+                    .durationType()
+                    .defaultValue(Duration.ofMillis(MIN_ABORT_GRACE_PERIOD_MS))
                     .withDescription(
-                            "The max time to wait for a deployment to become ready before aborting it, in milliseconds. Cannot be smaller than 2 minutes.");
+                            "The max time to wait in milliseconds for a deployment to become ready before aborting it. Cannot be smaller than 2 minutes.");
 
-    public static final ConfigOption<Integer> RECONCILIATION_RESCHEDULING_INTERVAL_MS =
-            operatorConfig("reconciliationReschedulingIntervalMs")
-                    .intType()
-                    .defaultValue(15000) // 15 seconds
+    public static final ConfigOption<Duration> RECONCILIATION_RESCHEDULING_INTERVAL =
+            operatorConfig("reconciliation.reschedule-interval")
+                    .durationType()
+                    .defaultValue(Duration.ofMillis(15000)) // 15 seconds
                     .withDescription(
                             "Configurable delay in milliseconds to use when the operator reschedules a reconciliation.");
 
-    public static final ConfigOption<Integer> DEPLOYMENT_DELETION_DELAY_MS =
-            operatorConfig("deploymentDeletionDelayMs")
-                    .intType()
-                    .defaultValue(0)
+    public static final ConfigOption<Duration> DEPLOYMENT_DELETION_DELAY =
+            operatorConfig("deployment-deletion.delay")
+                    .durationType()
+                    .defaultValue(Duration.ofMillis(0))
                     .withDescription(
-                            "Configurable delay before deleting a deployment after being marked done.");
+                            "Configurable delay in milliseconds before deleting a deployment after being marked done.");
 }
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentController.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentController.java
@@ -53,10 +53,10 @@
 import java.util.Map;
 import java.util.Optional;
 
-import static org.apache.flink.kubernetes.operator.api.spec.FlinkBlueGreenDeploymentConfigOptions.ABORT_GRACE_PERIOD_MS;
-import static org.apache.flink.kubernetes.operator.api.spec.FlinkBlueGreenDeploymentConfigOptions.DEPLOYMENT_DELETION_DELAY_MS;
+import static org.apache.flink.kubernetes.operator.api.spec.FlinkBlueGreenDeploymentConfigOptions.ABORT_GRACE_PERIOD;
+import static org.apache.flink.kubernetes.operator.api.spec.FlinkBlueGreenDeploymentConfigOptions.DEPLOYMENT_DELETION_DELAY;
 import static org.apache.flink.kubernetes.operator.api.spec.FlinkBlueGreenDeploymentConfigOptions.MIN_ABORT_GRACE_PERIOD_MS;
-import static org.apache.flink.kubernetes.operator.api.spec.FlinkBlueGreenDeploymentConfigOptions.RECONCILIATION_RESCHEDULING_INTERVAL_MS;
+import static org.apache.flink.kubernetes.operator.api.spec.FlinkBlueGreenDeploymentConfigOptions.RECONCILIATION_RESCHEDULING_INTERVAL;
 import static org.apache.flink.kubernetes.operator.controller.FlinkBlueGreenDeploymentUtils.getConfigOption;
 import static org.apache.flink.kubernetes.operator.controller.FlinkBlueGreenDeploymentUtils.instantStrToMillis;
 import static org.apache.flink.kubernetes.operator.controller.FlinkBlueGreenDeploymentUtils.millisToInstantStr;
@@ -94,20 +94,6 @@ public List<EventSource<?, FlinkBlueGreenDeployment>> prepareEventSources(
         return eventSources;
     }
 
-    //    @Override
-    //    public Map<String, EventSource> prepareEventSources(
-    //            EventSourceContext<FlinkBlueGreenDeployment> eventSourceContext) {
-    //        InformerConfiguration<FlinkDeployment> flinkDeploymentInformerConfig =
-    //                InformerConfiguration.from(FlinkDeployment.class, eventSourceContext)
-    //                        .withSecondaryToPrimaryMapper(Mappers.fromOwnerReference())
-    //                        .withNamespacesInheritedFromController(eventSourceContext)
-    //                        .followNamespaceChanges(true)
-    //                        .build();
-    //
-    //        return EventSourceInitializer.nameEventSources(
-    //                new InformerEventSource<>(flinkDeploymentInformerConfig, eventSourceContext));
-    //    }
-
     @Override
     public UpdateControl<FlinkBlueGreenDeployment> reconcile(
             FlinkBlueGreenDeployment bgDeployment, Context<FlinkBlueGreenDeployment> josdkContext)
@@ -186,7 +172,7 @@ private UpdateControl<FlinkBlueGreenDeployment> checkFirstDeployment(
     private static void setAbortTimestamp(
             FlinkBlueGreenDeployment bgDeployment,
             FlinkBlueGreenDeploymentStatus deploymentStatus) {
-        Integer abortGracePeriod = getConfigOption(bgDeployment, ABORT_GRACE_PERIOD_MS);
+        Long abortGracePeriod = getConfigOption(bgDeployment, ABORT_GRACE_PERIOD).toMillis();
         abortGracePeriod = Math.max(abortGracePeriod, minimumAbortGracePeriodMs);
         deploymentStatus.setAbortTimestamp(
                 millisToInstantStr(System.currentTimeMillis() + abortGracePeriod));
@@ -252,19 +238,17 @@ private UpdateControl<FlinkBlueGreenDeployment> canDelete(
             Context<FlinkBlueGreenDeployment> josdkContext,
             FlinkDeployment currentDeployment,
             FlinkBlueGreenDeploymentState nextState) {
-        int deploymentDeletionDelayMs =
-                Math.max(getConfigOption(bgDeployment, DEPLOYMENT_DELETION_DELAY_MS), 0);
+        long deploymentDeletionDelayMs =
+                Math.max(getConfigOption(bgDeployment, DEPLOYMENT_DELETION_DELAY).toMillis(), 0);
 
         long deploymentReadyTimestamp =
                 instantStrToMillis(deploymentStatus.getDeploymentReadyTimestamp());
 
         if (deploymentReadyTimestamp == 0) {
             LOG.info(
-                    "Deployment marked ready on "
-                            + System.currentTimeMillis()
-                            + ", rescheduling reconciliation in "
-                            + deploymentDeletionDelayMs
-                            + " ms.");
+                    "Deployment marked ready on {}, rescheduling reconciliation in {} ms.",
+                    System.currentTimeMillis(),
+                    deploymentDeletionDelayMs);
             deploymentStatus.setDeploymentReadyTimestamp(Instant.now().toString());
             return patchStatusUpdateControl(bgDeployment, deploymentStatus, null, null)
                     .rescheduleAfter(deploymentDeletionDelayMs);
@@ -277,7 +261,7 @@ private UpdateControl<FlinkBlueGreenDeployment> canDelete(
                     bgDeployment, deploymentStatus, josdkContext, currentDeployment, nextState);
         } else {
             long delay = deletionTs - System.currentTimeMillis();
-            LOG.info("Rescheduling reconciliation (to delete) in " + delay + " ms.");
+            LOG.info("Rescheduling reconciliation (to delete) in {} ms.", delay);
             return UpdateControl.<FlinkBlueGreenDeployment>noUpdate().rescheduleAfter(delay);
         }
     }
@@ -320,10 +304,9 @@ private UpdateControl<FlinkBlueGreenDeployment> shouldAbort(
             deploymentStatus.setBlueGreenState(previousState);
 
             LOG.warn(
-                    "Aborting deployment '"
-                            + deploymentName
-                            + "', rolling B/G deployment back to "
-                            + previousState);
+                    "Aborting deployment '{}', rolling B/G deployment back to {}",
+                    deploymentName,
+                    previousState);
 
             // If the current running FlinkDeployment is not in RUNNING/STABLE,
             // we flag this Blue/Green as FAILING
@@ -332,12 +315,7 @@ private UpdateControl<FlinkBlueGreenDeployment> shouldAbort(
         } else {
             // RETRY
             var delay = abortTimestamp - System.currentTimeMillis();
-            LOG.info(
-                    "Deployment '"
-                            + deploymentName
-                            + "' not ready yet, retrying in "
-                            + delay
-                            + " ms");
+            LOG.info("Deployment '{}' not ready yet, retrying in {} ms", deploymentName, delay);
             return patchStatusUpdateControl(bgDeployment, deploymentStatus, null, null)
                     .rescheduleAfter(delay);
         }
@@ -355,11 +333,9 @@ private UpdateControl<FlinkBlueGreenDeployment> deleteAndFinalize(
             return UpdateControl.<FlinkBlueGreenDeployment>noUpdate().rescheduleAfter(500);
         } else {
             LOG.info(
-                    "Finalizing deployment '"
-                            + bgDeployment.getMetadata().getName()
-                            + "' to "
-                            + nextState
-                            + " state");
+                    "Finalizing deployment '{}' to {} state",
+                    bgDeployment.getMetadata().getName(),
+                    nextState);
             deploymentStatus.setDeploymentReadyTimestamp(millisToInstantStr(0));
             deploymentStatus.setAbortTimestamp(millisToInstantStr(0));
             return patchStatusUpdateControl(
@@ -469,8 +445,11 @@ private UpdateControl<FlinkBlueGreenDeployment> initiateDeployment(
 
         setAbortTimestamp(bgDeployment, deploymentStatus);
 
-        var reconciliationReschedInterval =
-                Math.max(getConfigOption(bgDeployment, RECONCILIATION_RESCHEDULING_INTERVAL_MS), 0);
+        long reconciliationReschedInterval =
+                Math.max(
+                        getConfigOption(bgDeployment, RECONCILIATION_RESCHEDULING_INTERVAL)
+                                .toMillis(),
+                        0);
 
         return patchStatusUpdateControl(
                         bgDeployment, deploymentStatus, nextState, JobStatus.RECONCILING)
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentUtils.java
@@ -27,6 +27,7 @@
 
 import java.time.Instant;
 import java.util.List;
+import java.util.Map;
 
 /** Utility methods for the FlinkBlueGreenDeploymentController. */
 public class FlinkBlueGreenDeploymentUtils {
@@ -71,7 +72,12 @@ public static long instantStrToMillis(String instant) {
 
     public static <T> T getConfigOption(
             FlinkBlueGreenDeployment bgDeployment, ConfigOption<T> option) {
-        return Configuration.fromMap(bgDeployment.getSpec().getTemplate().getConfiguration())
-                .get(option);
+        Map<String, String> configuration = bgDeployment.getSpec().getTemplate().getConfiguration();
+
+        if (configuration == null) {
+            return option.defaultValue();
+        }
+
+        return Configuration.fromMap(configuration).get(option);
     }
 }
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentControllerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentControllerTest.java
@@ -59,9 +59,9 @@
 import java.util.Map;
 import java.util.UUID;
 
-import static org.apache.flink.kubernetes.operator.api.spec.FlinkBlueGreenDeploymentConfigOptions.ABORT_GRACE_PERIOD_MS;
-import static org.apache.flink.kubernetes.operator.api.spec.FlinkBlueGreenDeploymentConfigOptions.DEPLOYMENT_DELETION_DELAY_MS;
-import static org.apache.flink.kubernetes.operator.api.spec.FlinkBlueGreenDeploymentConfigOptions.RECONCILIATION_RESCHEDULING_INTERVAL_MS;
+import static org.apache.flink.kubernetes.operator.api.spec.FlinkBlueGreenDeploymentConfigOptions.ABORT_GRACE_PERIOD;
+import static org.apache.flink.kubernetes.operator.api.spec.FlinkBlueGreenDeploymentConfigOptions.DEPLOYMENT_DELETION_DELAY;
+import static org.apache.flink.kubernetes.operator.api.spec.FlinkBlueGreenDeploymentConfigOptions.RECONCILIATION_RESCHEDULING_INTERVAL;
 import static org.apache.flink.kubernetes.operator.api.utils.BaseTestUtils.SAMPLE_JAR;
 import static org.apache.flink.kubernetes.operator.api.utils.BaseTestUtils.TEST_DEPLOYMENT_NAME;
 import static org.apache.flink.kubernetes.operator.api.utils.BaseTestUtils.TEST_NAMESPACE;
@@ -183,9 +183,9 @@ public void verifyFailureDuringTransition(FlinkVersion flinkVersion) throws Exce
         var reconciliationReschedulingIntervalMs = 5000;
         Map<String, String> configuration =
                 blueGreenDeployment.getSpec().getTemplate().getConfiguration();
-        configuration.put(ABORT_GRACE_PERIOD_MS.key(), String.valueOf(abortGracePeriodMs));
+        configuration.put(ABORT_GRACE_PERIOD.key(), String.valueOf(abortGracePeriodMs));
         configuration.put(
-                RECONCILIATION_RESCHEDULING_INTERVAL_MS.key(),
+                RECONCILIATION_RESCHEDULING_INTERVAL.key(),
                 String.valueOf(reconciliationReschedulingIntervalMs));
 
         var rs = executeBasicDeployment(flinkVersion, blueGreenDeployment, false);
@@ -526,7 +526,7 @@ private void simulateChangeInSpec(
 
         if (customDeletionDelayMs > 0) {
             template.getConfiguration()
-                    .put(DEPLOYMENT_DELETION_DELAY_MS.key(), String.valueOf(customDeletionDelayMs));
+                    .put(DEPLOYMENT_DELETION_DELAY.key(), String.valueOf(customDeletionDelayMs));
         }
 
         FlinkDeploymentSpec spec = template.getSpec();
@@ -648,10 +648,10 @@ private static FlinkBlueGreenDeploymentSpec getTestFlinkDeploymentSpec(FlinkVers
                         .build();
 
         Map<String, String> configuration = new HashMap<>();
-        configuration.put(ABORT_GRACE_PERIOD_MS.key(), "1");
-        configuration.put(RECONCILIATION_RESCHEDULING_INTERVAL_MS.key(), "500");
+        configuration.put(ABORT_GRACE_PERIOD.key(), "1");
+        configuration.put(RECONCILIATION_RESCHEDULING_INTERVAL.key(), "500");
         configuration.put(
-                DEPLOYMENT_DELETION_DELAY_MS.key(), String.valueOf(DEFAULT_DELETION_DELAY_VALUE));
+                DEPLOYMENT_DELETION_DELAY.key(), String.valueOf(DEFAULT_DELETION_DELAY_VALUE));
 
         var flinkDeploymentTemplateSpec =
                 FlinkDeploymentTemplateSpec.builder()
