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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
