Dennis-Mircea commented on code in PR #1197:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/1197#discussion_r4027085536
##########
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:
Should we wait for the new JOSDK patch to be adopted as part of the operator
or is it fine to go with the workaround that @spuru9 suggested? cc @gyfora
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventUtils.java:
##########
@@ -271,7 +271,7 @@ private static boolean labelCheck(
private static Optional<Event> createOrReplaceEvent(KubernetesClient
client, Event event) {
try {
- Event createdEvent = client.resource(event).createOrReplace();
+ Event createdEvent = client.resource(event).createOr(r ->
r.unlock().update());
Review Comment:
The thing is that, having just `update` in place, this will perform an
optimistic-lock PUT, and if multiple concurrent updates happen at the same
time, it will result in 409 and it will fail the reconciliation cycle.
The old `createOrReplace` was a forced replace, which is what we want for
best-effort events, and `unlock()` was meant to keep that.
Looking again at this point from a different perspective, the `unlock()`
only runs on the conflict branch, after the `create` returns, but `create` can
fail with a 500 when a POST is executed with the same resourceVersion (see
https://github.com/kubernetes/kubernetes/blob/master/staging/src/k8s.io/apiserver/pkg/storage/errors.go#L30).
To resolve this, I used the fabric8 migration FAQ recommendation presented
here:
https://github.com/fabric8io/kubernetes-client/blob/main/doc/FAQ.md#alternatives-to-createorreplace-and-replace.
I think we should all good with this approach.
##########
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:
Good suggestion, I've added it.
--
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]
