loserwang1024 commented on code in PR #3649:
URL: https://github.com/apache/fluss/pull/3649#discussion_r3601662597
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorLeaderElection.java:
##########
@@ -186,18 +140,138 @@ public void notLeader() {
public void close() {
LOG.info("Closing LeaderLatch for server {}.", serverId);
- if (leaderLatch != null) {
+ if (closing.compareAndSet(false, true)) {
try {
leaderLatch.close();
} catch (Exception e) {
LOG.error("Failed to close LeaderLatch for server {}.",
serverId, e);
}
+
+ // Events submitted after closing starts are ignored by their
executor-side check.
+ // Since the executor is single-threaded, this task runs after all
leadership work
+ // already queued before close and completes only after leader
cleanup has finished.
+ leaderCallbackExecutor.execute(
+ () -> {
+ try {
+ boolean cleanupRequired = state == State.LEADER;
+ state = State.CLOSED;
+ if (cleanupRequired) {
+ cleanupLeaderServices(null);
+ }
+ } finally {
+ closeFuture.complete(null);
+ }
+ });
}
- leaderCallbackExecutor.shutdownNow();
+ closeFuture.join();
+ leaderCallbackExecutor.shutdown();
+ awaitCallbackExecutorTermination();
}
public boolean isLeader() {
- return this.isLeader.get();
+ return !closing.get() && state == State.LEADER;
+ }
+
+ private void submitLeadershipEvent(boolean leader, Runnable
initLeaderServices) {
Review Comment:
done
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorLeaderElection.java:
##########
@@ -186,18 +140,138 @@ public void notLeader() {
public void close() {
LOG.info("Closing LeaderLatch for server {}.", serverId);
- if (leaderLatch != null) {
+ if (closing.compareAndSet(false, true)) {
try {
leaderLatch.close();
} catch (Exception e) {
LOG.error("Failed to close LeaderLatch for server {}.",
serverId, e);
}
+
+ // Events submitted after closing starts are ignored by their
executor-side check.
+ // Since the executor is single-threaded, this task runs after all
leadership work
+ // already queued before close and completes only after leader
cleanup has finished.
+ leaderCallbackExecutor.execute(
+ () -> {
+ try {
+ boolean cleanupRequired = state == State.LEADER;
+ state = State.CLOSED;
+ if (cleanupRequired) {
+ cleanupLeaderServices(null);
+ }
+ } finally {
+ closeFuture.complete(null);
+ }
+ });
}
- leaderCallbackExecutor.shutdownNow();
+ closeFuture.join();
+ leaderCallbackExecutor.shutdown();
+ awaitCallbackExecutorTermination();
}
public boolean isLeader() {
- return this.isLeader.get();
+ return !closing.get() && state == State.LEADER;
+ }
+
+ private void submitLeadershipEvent(boolean leader, Runnable
initLeaderServices) {
+ if (closing.get()) {
+ return;
+ }
+ try {
+ leaderCallbackExecutor.execute(
+ () -> {
+ if (closing.get() || state == State.CLOSED) {
+ return;
+ }
+ if (leader) {
+ becomeLeader(initLeaderServices);
+ } else {
+ becomeStandby();
+ }
+ });
+ } catch (RejectedExecutionException e) {
+ if (!closing.get()) {
+ throw e;
+ }
+ }
+ }
+
+ private void becomeLeader(Runnable initLeaderServices) {
+ if (closing.get() || state == State.LEADER || state == State.CLOSED) {
+ return;
+ }
+
+ LOG.info("Coordinator server {} has become the leader.", serverId);
+ Throwable initializationFailure = null;
+ try {
+ initLeaderServices.run();
+ } catch (CoordinatorEpochFencedException e) {
+ LOG.warn(
+ "Coordinator server {} has been fenced and not become
leader successfully.",
+ serverId);
+ initializationFailure = e;
+ } catch (Exception e) {
+ LOG.error("Failed to initialize leader services for server {}",
serverId, e);
+ initializationFailure = e;
+ }
+
+ if (initializationFailure == null && !closing.get()) {
+ state = State.LEADER;
+ } else {
+ cleanupLeaderServices(initializationFailure);
Review Comment:
You are right. This behavior comes from the existing strategy.
I also considered releasing the current leadership when leader
initialization fails, for example by closing the current LeaderLatch. However,
that alone would be problematic for a single-coordinator deployment, because
there may be no other candidate to take over. To handle this correctly, we
likely need to release the current election leadership and then rejoin the
election, probably with some backoff to avoid a tight retry loop.
That recovery behavior is outside the scope of this PR, so I think we should
handle it in a separate follow-up. For this PR, I added a
regression/reproduction test to document the current behavior:
`org.apache.fluss.server.coordinator.CoordinatorLeaderElectionTest#testInitializationFailureKeepsLeaderLatchAndBlocksReElection`.
--
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]
