HDDS-14989. Delay follower SCM DN server start until Ratis log catch-up.

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/AbstractContainerReportHandler.java

@@ -253,7 +253,7 @@ private boolean updateContainerState(final DatanodeDetails datanode,
       // If the state of a container is OPEN and a replica is in different state, finalize the container.
       if (replica.getState() != State.OPEN) {
         getLogger().info("FINALIZE (i.e. CLOSING) {}", detailsForLogging);
-        containerManager.updateContainerState(containerId, LifeCycleEvent.FINALIZE);
+        updateContainerState(containerId, LifeCycleEvent.FINALIZE);
       }
       return false;
     case CLOSING:
@@ -264,7 +264,7 @@ private boolean updateContainerState(final DatanodeDetails datanode,
       // If the replica is in QUASI_CLOSED state, move the container to QUASI_CLOSED state.
       if (replica.getState() == State.QUASI_CLOSED) {
         getLogger().info("QUASI_CLOSE {}", detailsForLogging);
-        containerManager.updateContainerState(containerId, LifeCycleEvent.QUASI_CLOSE);
+        updateContainerState(containerId, LifeCycleEvent.QUASI_CLOSE);
         return false;
       }
 
@@ -288,7 +288,7 @@ private boolean updateContainerState(final DatanodeDetails datanode,
           return true;
         }
         getLogger().info("CLOSE {}", detailsForLogging);
-        containerManager.updateContainerState(containerId, LifeCycleEvent.CLOSE);
+        updateContainerState(containerId, LifeCycleEvent.CLOSE);
       }
       return false;
     case QUASI_CLOSED:
@@ -301,7 +301,7 @@ private boolean updateContainerState(final DatanodeDetails datanode,
           return true;
         }
         getLogger().info("FORCE_CLOSE for {}", detailsForLogging);
-        containerManager.updateContainerState(containerId, LifeCycleEvent.FORCE_CLOSE);
+        updateContainerState(containerId, LifeCycleEvent.FORCE_CLOSE);
       }
       return false;
     case CLOSED:
@@ -362,6 +362,24 @@ private boolean updateContainerState(final DatanodeDetails datanode,
     }
   }
 
+  /**
+   * Apply a container lifecycle state transition, but only on the leader SCM.
+   * On a follower the underlying {@code containerManager.updateContainerState}
+   * is a Ratis write and would throw {@code NotLeaderException}, which would
+   * abort {@code processContainerReplica} and skip recording the replica
+   * location. Skipping the state change on a follower is safe: the leader
+   * drives the transition and it replicates back via the Ratis log.
+   */
+  private void updateContainerState(ContainerID containerID, LifeCycleEvent event)
+      throws InvalidStateTransitionException, IOException {
+    if (scmContext.isLeader()) {
+      containerManager.updateContainerState(containerID, event);
+    } else {
+      getLogger().debug("Skipping updateContainerState on non-leader SCM, container {} event {}",
+          containerID, event);
+    }
+  }
+
   /**
    * Helper method to verify that the replica's bcsId matches the container's in SCM.
    *

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMStateMachine.java

@@ -32,7 +32,6 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
 import org.apache.hadoop.hdds.protocol.proto.SCMRatisProtocol.RequestType;
 import org.apache.hadoop.hdds.scm.block.DeletedBlockLog;
 import org.apache.hadoop.hdds.scm.block.DeletedBlockLogImpl;
@@ -86,9 +85,14 @@ public class SCMStateMachine extends BaseStateMachine {
   private DBCheckpoint installingDBCheckpoint = null;
   private List<ManagedSecretKey> installingSecretKeys = null;
 
-  private AtomicLong currentLeaderTerm = new AtomicLong(-1L);
   private AtomicBoolean isStateMachineReady = new AtomicBoolean();
 
+  // The leader's committed index captured when this SCM (re)joins as a
+  // follower. Catch-up is measured against this fixed target rather than the
+  // leader's live commit index, which on a busy cluster keeps advancing and
+  // would never be reached. Set only while not yet ready; -1 means uncaptured.
+  private volatile long leaderCommitIndexOnStart = -1L;
+
   public SCMStateMachine(final StorageContainerManager scm,
       SCMHADBTransactionBuffer buffer) {
     this.scm = scm;
@@ -181,6 +185,11 @@ public CompletableFuture<Message> applyTransaction(
       final TermIndex appliedTermIndex = TermIndex.valueOf(trx.getLogEntry());
       transactionBuffer.updateLatestTrxInfo(TransactionInfo.valueOf(appliedTermIndex));
       updateLastAppliedTermIndex(appliedTermIndex);
+
+      // A restarted follower may catch up by applying data-carrying entries
+      // here rather than through notifyTermIndexUpdated, so check for catch-up
+      // in both places. No-op once the datanode protocol server has started.
+      tryStartDNServerAndRefreshSafeMode();
     } catch (Exception ex) {
       applyTransactionFuture.completeExceptionally(ex);
       ExitUtils.terminate(1, ex.getMessage(), ex, StateMachine.LOG);
@@ -284,34 +293,40 @@ public void notifyLeaderChanged(RaftGroupMemberId groupMemberId,
       return;
     }
 
-    currentLeaderTerm.set(scm.getScmHAManager().getRatisServer().getDivision()
-        .getInfo().getCurrentTerm());
-
-    if (isStateMachineReady.compareAndSet(false, true)) {
-      // refresh and validate safe mode rules if it can exit safe mode
-      // if being leader, all previous term transactions have been applied
-      // if other states, just refresh safe mode rules, and transaction keeps flushing from leader
-      // and does not depend on pending transactions.
-      scm.getScmSafeModeManager().refreshAndValidate();
-    }
-
-    if (!groupMemberId.getPeerId().equals(newLeaderId)) {
+    final boolean isLeader = groupMemberId.getPeerId().equals(newLeaderId);
+
+    if (!isLeader) {
+      // Follower: capture the (possibly new) leader's current committed index
+      // as the fixed catch-up target, then start the datanode protocol server
+      // if we are already caught up with it; otherwise applyTransaction /
+      // notifyTermIndexUpdated start it as catch-up completes. Set it always:
+      // getLeaderCommitIndex() returns -1 when the leader is not known yet,
+      // which isFollowerCaughtUp() treats as uncaptured and re-reads later.
+      if (!isStateMachineReady.get()) {
+        leaderCommitIndexOnStart = getLeaderCommitIndex();
+      }
+      tryStartDNServerAndRefreshSafeMode();
       String message = "Leader changed to " + newLeaderId +
           ", current SCM " + scm.getScmId() + " is still follower.";
       LOG.info(message);
       addRatisEvent(message);
       return;
     }
 
+    long currentTerm = scm.getScmHAManager().getRatisServer().getDivision()
+        .getInfo().getCurrentTerm();
     String message = "current SCM " + scm.getScmId() +
-        " becomes leader of term " + currentLeaderTerm;
+        " becomes leader of term " + currentTerm;
     LOG.info(message);
     addRatisEvent(message);
 
-    scm.getScmContext().updateLeaderAndTerm(true,
-        currentLeaderTerm.get());
+    scm.getScmContext().updateLeaderAndTerm(true, currentTerm);
     scm.getSequenceIdGen().invalidateBatch();
 
+    // isLeader() is now true -> start the datanode protocol server for the new
+    // leader (a leader has applied all committed entries) and refresh safe mode.
+    tryStartDNServerAndRefreshSafeMode();
+
     try {
       transactionBuffer.flush();
     } catch (Exception ex) {
@@ -386,16 +401,93 @@ 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 captured when it (re)joined, i.e. all transactions the
+   * leader had committed at that point have been replayed. Comparing against a
+   * fixed target avoids chasing the leader's ever-advancing live commit index.
+   */
+  private boolean isFollowerCaughtUp() {
+    try {
+      long target = leaderCommitIndexOnStart;
+      if (target < 0) {
+        // Not captured at leader-change time yet; capture the leader's current
+        // commit index once here so we still compare against a fixed target.
+        target = getLeaderCommitIndex();
+        if (target < 0) {
+          // Normal transient condition during startup/catch-up; this is polled
+          // from multiple callbacks, so keep it at DEBUG to avoid log flooding.
+          LOG.debug("Leader commit index not available yet");
+          return false;
+        }
+        leaderCommitIndexOnStart = target;
+      }
+
+      long lastAppliedIndex = scm.getScmHAManager().getRatisServer()
+          .getDivision().getInfo().getLastAppliedIndex();
+      boolean caughtUp = lastAppliedIndex >= target;
+      if (caughtUp) {
+        LOG.info("Follower caught up with leader: lastAppliedIndex={}, leaderCommitOnStart={}",
+            lastAppliedIndex, target);
+      } else {
+        LOG.debug("Follower not caught up: lastAppliedIndex={}, leaderCommitOnStart={}",
+            lastAppliedIndex, target);
+      }
+      return caughtUp;
+    } catch (Exception e) {
+      LOG.warn("Failed to check follower catch-up status", e);
+      return false;
+    }
+  }
+
+  /**
+   * @return the leader's current committed index as seen by this SCM, or -1 if
+   * the leader or its commit info is not available yet.
+   */
+  private long getLeaderCommitIndex() {
+    RaftServer.Division division = scm.getScmHAManager()
+        .getRatisServer().getDivision();
+    RaftPeerId leaderId = division.getInfo().getLeaderId();
+    if (leaderId != null) {
+      for (RaftProtos.CommitInfoProto info : division.getCommitInfos()) {
+        if (info.getServer().getId().equals(leaderId.toByteString())) {
+          return info.getCommitIndex();
+        }
+      }
+    }
+    return -1L;
+  }
+
   public boolean getIsStateMachineReady() {
     return isStateMachineReady.get();
   }

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java

@@ -1580,8 +1580,13 @@ public void start() throws IOException {
     }
     getBlockProtocolServer().start();
 
-    // start datanode protocol server
-    getDatanodeProtocolServer().start();
+    // In HA mode, defer starting the datanode protocol server until the SCM
+    // state machine has caught up with the leader's committed log entries
+    // (see SCMStateMachine#tryStartDNServerAndRefreshSafeMode). In non-HA mode
+    // there is no Ratis state machine, so start it here as before.
+    if (!scmStorageConfig.isSCMHAEnabled()) {
+      getDatanodeProtocolServer().start();
+    }
     if (getSecurityProtocolServer() != null) {
       getSecurityProtocolServer().start();
       persistSCMCertificates();