ivandika3 commented on code in PR #10059:
URL: https://github.com/apache/ozone/pull/10059#discussion_r3121480790
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMStateMachine.java:
##########
@@ -369,19 +366,100 @@ public void notifyTermIndexUpdated(long term, long
index) {
if (transactionBuffer != null) {
transactionBuffer.updateLatestTrxInfo(TransactionInfo.valueOf(term,
index));
}
+ }
+
+ public boolean getIsStateMachineReady() {
+ return isStateMachineReady.get();
+ }
- if (currentLeaderTerm.get() == term) {
- // This means after a restart, all pending transactions have been
applied.
+ /**
+ * Start the DN protocol server and trigger safe mode re-evaluation.
+ *
+ * <p>In HA mode the DN server is deliberately not started during
+ * {@link org.apache.hadoop.hdds.scm.server.StorageContainerManager#start()}.
+ * Instead it is deferred until the SCM state machine has caught up with
+ * the leader's committed log entries, so that DN heartbeats are processed
+ * against the latest container/pipeline state rather than a stale snapshot.
+ *
+ * <p>The method is guarded by {@code isStateMachineReady} (CAS) to ensure
+ * the non-idempotent {@code DatanodeProtocolServer.start()} is invoked
+ * exactly once.
+ */
+ 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);
}
}
- public boolean getIsStateMachineReady() {
- return isStateMachineReady.get();
+ /**
+ * Check whether this follower's state machine has caught up with the
+ * leader's committed log entries.
+ * @return true if {@code lastAppliedIndex >= leaderCommitIndex}
+ */
+ 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();
+ boolean caughtUp = lastAppliedIndex >= leaderCommit;
Review Comment:
Let's apply the fix where we need to take into account the metadata entry.
--
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]
