Re: [PR] [hotfix] Replace deprecated Flink, JOSDK, and fabric8 APIs [flink-kubernetes-operator]

["spuru9 (via GitHub)" <gi...@apache.org>]

spuru9 commented on code in PR #1197:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1197#discussion_r3911391552


##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java:
##########
@@ -313,20 +313,30 @@ private static LeaderElectionConfiguration getLeaderElectionConfig(Configuration
             return null;
         }
 
-        return new LeaderElectionConfiguration(
-                conf.getOptional(
-                                KubernetesOperatorConfigOptions.OPERATOR_LEADER_ELECTION_LEASE_NAME)
-                        .orElseThrow(
-                                () ->
-                                        new IllegalConfigurationException(
-                                                KubernetesOperatorConfigOptions
-                                                                .OPERATOR_LEADER_ELECTION_LEASE_NAME
-                                                                .key()
-                                                        + " must be defined when operator leader election is enabled.")),
-                null,
-                conf.get(KubernetesOperatorConfigOptions.OPERATOR_LEADER_ELECTION_LEASE_DURATION),
-                conf.get(KubernetesOperatorConfigOptions.OPERATOR_LEADER_ELECTION_RENEW_DEADLINE),
-                conf.get(KubernetesOperatorConfigOptions.OPERATOR_LEADER_ELECTION_RETRY_PERIOD));
+        return LeaderElectionConfigurationBuilder.aLeaderElectionConfiguration(

Review Comment:
   In JOSDK 5.5.0, `LeaderElectionConfigurationBuilder.build()` is implemented as:
   ```java
       public LeaderElectionConfiguration build() {
           return buildForTest(false);
       }
   ```
   
   This hardcodes exitOnStopLeading = false (the builder's Javadoc notes false should only be used for tests).
   
     Previously, the deprecated constructor new LeaderElectionConfiguration(...) set exitOnStopLeading = true. When exitOnStopLeading is true, if the operator loses its leadership lease, LeaderElectionManager.stopLeading() executes System.exit(1) so Kubernetes restarts the pod and allows a standby replica to take over cleanly.
   
   With .build(), exitOnStopLeading becomes false, so stopLeading() will merely log "Stopped leading, configured not to exit" and keep running without exiting. This can cause split-brain or stuck reconciliation threads during HA failovers.
   
   Please use .buildForTest(true) here so that production HA failover semantics (System.exit(1) on lost lease) are preserved.



##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkUtils.java:
##########
@@ -258,7 +258,11 @@ public static void deleteJobGraphInKubernetesHA(
             }
         }
         if (shouldUpdate) {
-            kubernetesClient.resourceList(configMaps).inNamespace(namespace).createOrReplace();
+            kubernetesClient
+                    .resourceList(configMaps)
+                    .inNamespace(namespace)
+                    .resources()
+                    .forEach(Resource::update);

Review Comment:
   `configMaps` contains all HA ConfigMaps for the cluster (e.g. dispatch leader, JM leader, etc.). Calling `.resources().forEach(Resource::update)` sends an HTTP `PUT` update for **every single ConfigMap** in the list, even those that were untouched.
       
       1. It issues unnecessary `PUT` requests for unmodified ConfigMaps.
       2. If any untouched ConfigMap (such as an active leader lease ConfigMap) was concurrently modified in the cluster, its `resourceVersion` will be stale and `Resource::update` will throw a `409 Conflict` exception, aborting the cleanup.
       
   Suggestion: collect only the modified ConfigMaps in the loop above and update only those:
   ```
               List<ConfigMap> modifiedConfigMaps = new ArrayList<>();
               for (ConfigMap configMap : configMaps.getItems()) {
                   if (configMap.getData() == null || configMap.getData().isEmpty()) {
                       continue;
                   }
                   if (configMap.getData().entrySet().removeIf(FlinkUtils::isJobGraphKey)) {
                       modifiedConfigMaps.add(configMap);
                       LOG.info("Job graph in ConfigMap {} is deleted", configMap.getMetadata().getName());
                   }
               }
               modifiedConfigMaps.forEach(cm -> kubernetesClient.resource(cm).inNamespace(namespace).update());
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org