ArafatKhan2198 commented on code in PR #10617:
URL: https://github.com/apache/ozone/pull/10617#discussion_r3542554736
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMStateMachine.java:
##########
@@ -380,13 +385,69 @@ public void notifyTermIndexUpdated(long term, long index)
{
}
}
- if (currentLeaderTerm.get() == term) {
- // This means after a restart, all pending transactions have been
applied.
+ // As committed entries are applied (e.g. a restarted follower catching
up),
+ // start the datanode protocol server once we are caught up with the
leader's
+ // committed index. No-op once the server has already been started.
+ tryStartDNServerAndRefreshSafeMode();
+ }
+
+ /**
+ * Start the DatanodeProtocolServer and re-evaluate safe-mode rules, but only
+ * when this SCM is safe to accept datanode reports: it is the leader, or it
+ * is a follower whose state machine has caught up with the leader's
committed
+ * log. Guarded by {@code isStateMachineReady} (CAS) so the non-idempotent
+ * {@code DatanodeProtocolServer.start()} runs exactly once.
+ *
+ * <p>In HA mode {@link StorageContainerManager#start()} deliberately does
not
+ * start the datanode protocol server; it is deferred to here so datanode
+ * container reports are processed against the up-to-date container/pipeline
+ * state rather than a stale, mid-replay snapshot.
+ */
+ private void tryStartDNServerAndRefreshSafeMode() {
+ if (isStateMachineReady.get()) {
+ return;
+ }
+ if (scm.getScmContext().isLeader() || isFollowerCaughtUp()) {
if (isStateMachineReady.compareAndSet(false, true)) {
- // Refresh Safemode rules state if not already done.
+ scm.getDatanodeProtocolServer().start();
scm.getScmSafeModeManager().refreshAndValidate();
}
- currentLeaderTerm.set(-1L);
+ }
+ }
+
+ /**
+ * @return true if this follower's last applied index has reached the
leader's
+ * committed index, i.e. all committed transactions have been replayed.
+ */
+ private boolean isFollowerCaughtUp() {
+ try {
+ RaftServer.Division division = scm.getScmHAManager()
+ .getRatisServer().getDivision();
+ DivisionInfo divisionInfo = division.getInfo();
+ long lastAppliedIndex = divisionInfo.getLastAppliedIndex();
+
+ RaftPeerId leaderId = divisionInfo.getLeaderId();
+ if (leaderId != null) {
+ for (RaftProtos.CommitInfoProto info : division.getCommitInfos()) {
+ if (info.getServer().getId().equals(leaderId.toByteString())) {
+ long leaderCommit = info.getCommitIndex();
Review Comment:
I now capture the leader's commit index once in `notifyLeaderChanged` (only
while `isStateMachineReady` is still false) into a new
`leaderCommitIndexOnStart` field, and `isFollowerCaughtUp()` compares
`lastAppliedIndex` against that fixed target instead of the live commit index.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
