XComp commented on code in PR #24132:
URL: https://github.com/apache/flink/pull/24132#discussion_r1469394774
##########
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElector.java:
##########
@@ -86,12 +106,33 @@ public KubernetesLeaderElector(
newLeader,
leaderConfig.getConfigMapName())))
.build();
- internalLeaderElector =
- new LeaderElector(kubernetesClient, leaderElectionConfig,
executorService);
+ this.executorService = executorService;
+
LOG.info(
- "Create KubernetesLeaderElector {} with lock identity {}.",
- leaderConfig.getConfigMapName(),
- leaderConfig.getLockIdentity());
+ "Create KubernetesLeaderElector on lock {}.",
+ leaderElectionConfig.getLock().describe());
+ }
+
+ @GuardedBy("lock")
+ private void resetInternalLeaderElector() {
+ cancelCurrentLeaderElectionSession();
+
+ currentLeaderElectionSession =
+ currentLeaderElectionSession.thenCompose(
Review Comment:
true, I was wrong in my assumption that `cancel` would cancel also future
chains that are created by `thenCompose`. But tbh, the `thenCompose` is not
really necessary because we're cancelling the previous leadership session
before re-re-initiating another session all being guarded by a lock. I fixed it
and added another test case to cover this behavior. :+1:
--
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-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
