You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2024/12/10 08:15:06 UTC
(flink) 01/03: [hotfix][runtime] hasLeadership does not need to be called to protect the confirmLeadership call because the check happens internally
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch release-1.20
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 494cf93bc009158629ab5e944bbf55c1c0cfbe70
Author: Matthias Pohl <mp...@confluent.io>
AuthorDate: Mon Dec 2 15:14:16 2024 +0100
[hotfix][runtime] hasLeadership does not need to be called to protect the confirmLeadership call because the check happens internally
---
.../flink/runtime/dispatcher/runner/DefaultDispatcherRunner.java | 7 ++-----
.../flink/runtime/jobmaster/JobMasterServiceLeadershipRunner.java | 3 +--
.../apache/flink/runtime/leaderelection/DefaultLeaderElection.java | 3 ++-
.../org/apache/flink/runtime/leaderelection/LeaderElection.java | 2 ++
.../apache/flink/runtime/leaderelection/TestingLeaderElection.java | 4 +++-
5 files changed, 10 insertions(+), 9 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/DefaultDispatcherRunner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/DefaultDispatcherRunner.java
index d034e56dd4f..ce6579a67b9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/DefaultDispatcherRunner.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/DefaultDispatcherRunner.java
@@ -177,12 +177,9 @@ public final class DefaultDispatcherRunner implements DispatcherRunner, LeaderCo
newDispatcherLeaderProcess
.getLeaderAddressFuture()
.thenAccept(
- leaderAddress -> {
- if (leaderElection.hasLeadership(leaderSessionID)) {
+ leaderAddress ->
leaderElection.confirmLeadership(
- leaderSessionID, leaderAddress);
- }
- }));
+ leaderSessionID, leaderAddress)));
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunner.java
index ee7112cbbb4..256088460ad 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunner.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunner.java
@@ -356,8 +356,7 @@ public class JobMasterServiceLeadershipRunner implements JobManagerRunner, Leade
FutureUtils.assertNoException(
leaderAddressFuture.thenAccept(
address ->
- runIfValidLeader(
- leaderSessionId,
+ runIfStateRunning(
() -> {
LOG.debug("Confirm leadership {}.", leaderSessionId);
leaderElection.confirmLeadership(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElection.java b/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElection.java
index c1524ef01d4..06d6b62fb65 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElection.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElection.java
@@ -78,7 +78,8 @@ class DefaultLeaderElection implements LeaderElection {
/**
* Confirms the leadership with the {@code leaderSessionID} and {@code leaderAddress} for
- * the {@link LeaderContender} that is associated with the {@code componentId}.
+ * the {@link LeaderContender} that is associated with the {@code componentId}. The
+ * information is only propagated to the HA backend if the leadership is still acquired.
*/
abstract void confirmLeadership(
String componentId, UUID leaderSessionID, String leaderAddress);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/LeaderElection.java b/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/LeaderElection.java
index c78621d6865..0dc4ff55621 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/LeaderElection.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/LeaderElection.java
@@ -33,6 +33,8 @@ public interface LeaderElection extends AutoCloseable {
* Confirms that the {@link LeaderContender} has accepted the leadership identified by the given
* leader session id. It also publishes the leader address under which the leader is reachable.
*
+ * <p>The data is only confirmed if the leadership is still acquired.
+ *
* <p>The intention of this method is to establish an order between setting the new leader
* session ID in the {@link LeaderContender} and publishing the new leader session ID and the
* related leader address to the leader retrieval services.
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/TestingLeaderElection.java b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/TestingLeaderElection.java
index 384ef01ba18..cd59be89732 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/TestingLeaderElection.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/TestingLeaderElection.java
@@ -66,7 +66,9 @@ public class TestingLeaderElection implements LeaderElection {
@Override
public synchronized void confirmLeadership(UUID leaderSessionID, String leaderAddress) {
- if (confirmationFuture != null && !confirmationFuture.isDone()) {
+ if (leaderSessionID.equals(this.issuedLeaderSessionId)
+ && confirmationFuture != null
+ && !confirmationFuture.isDone()) {
confirmationFuture.complete(LeaderInformation.known(leaderSessionID, leaderAddress));
}
}