swuferhong commented on code in PR #3649:
URL: https://github.com/apache/fluss/pull/3649#discussion_r3585519636
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorLeaderElection.java:
##########
@@ -194,10 +172,51 @@ public void close() {
}
}
- leaderCallbackExecutor.shutdownNow();
+ // LeaderLatch notifications are asynchronous. Explicitly trigger the
same idempotent
+ // cleanup path to guarantee a clean shutdown even if notLeader() has
not arrived yet.
+ triggerCleanupLeaderServices();
+
+ leaderCallbackExecutor.shutdown();
+ try {
+ if (!leaderCallbackExecutor.awaitTermination(60,
TimeUnit.SECONDS)) {
+ LOG.warn(
+ "Coordinator leader callbacks for server {} did not
terminate within 60s.",
+ serverId);
+ leaderCallbackExecutor.shutdownNow();
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ leaderCallbackExecutor.shutdownNow();
+ }
}
public boolean isLeader() {
return this.isLeader.get();
}
+
+ private void triggerCleanupLeaderServices() {
+ if (isLeader.compareAndSet(true, false)) {
Review Comment:
`isLeader` is only set to true after `initLeaderServices.run()` completes.
If `close()` is called before initialization completes, this problem also will
be happen. The implementation should explicitly track `initialization/closing`
state instead of using `isLeader` as both readiness state and cleanup ownership
state?
--
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]
