HDDS-16798. Stale or Dead Datanodes whose software is not upgraded may still communicate with a finalized SCM

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

@@ -408,6 +408,16 @@ void processNodeReport(DatanodeDetails datanodeDetails,
    */
   void processVersionReport(DatanodeDetails datanodeDetails, DatanodeVersionProto versionReport);
 
+  /**
+   * Returns true if the datanode should be blocked from joining the cluster due to a version
+   * incompatibility with this SCM. See {@code SCMNodeManager.shouldFenceDatanode} for the full
+   * set of conditions. Returns false by default for test implementations that do not enforce versioning.
+   */
+  default boolean shouldFenceDatanode(DatanodeDetails datanodeDetails,
+      DatanodeVersionProto versionReport) {
+    return false;
+  }
+
   /**
    * Get the number of commands of the given type queued on the datanode at the
    * last heartbeat. If the Datanode has not reported information for the given

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

@@ -2112,7 +2112,8 @@ public void removeNode(DatanodeDetails datanodeDetails) throws NodeNotFoundExcep
     }
   }
 
-  protected boolean shouldFenceDatanode(DatanodeDetails dnDetails, DatanodeVersionProto versionReport) {
+  @Override
+  public boolean shouldFenceDatanode(DatanodeDetails dnDetails, DatanodeVersionProto versionReport) {
     if (versionReport == null || !versionReport.hasSoftwareVersion() || !versionReport.hasApparentVersion()) {
       LOG.error("Datanode {} did not report its version. Not allowing it to join the cluster.", dnDetails);
       return true;

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

@@ -42,6 +42,7 @@
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMHeartbeatRequestProto;
 import org.apache.hadoop.hdds.scm.node.NodeManager;
+import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException;
 import org.apache.hadoop.hdds.server.events.EventPublisher;
 import org.apache.hadoop.hdds.server.events.IEventInfo;
 import org.apache.hadoop.ozone.protocol.commands.ReregisterCommand;
@@ -92,9 +93,22 @@ public List<SCMCommand<?>> dispatch(SCMHeartbeatRequestProto heartbeat) {
 
       commands = nodeManager.getCommandQueue(dnID);
 
+    } else if (isUnhealthyWithIncompatibleVersion(datanodeDetails, heartbeat)) {
+      // Stale and dead nodes must pass a version compatibility check before their heartbeat
+      // timestamp is updated. Without this gate a node that went unhealthy during a rolling
+      // upgrade could resurrect at an incompatible software version. If the check fails, ask
+      // the node to re-register so the full registration fence runs with current version info.
+      LOG.info("Node {} is not healthy and reported an incompatible version {}. Asking to re-register.",
+          datanodeDetails, heartbeat.getDatanodeVersion());
+      nodeManager.addDatanodeCommand(datanodeDetails.getID(), new ReregisterCommand());
+
+      commands = nodeManager.getCommandQueue(datanodeDetails.getID());
+
     } else {
 
-      // Process the last known version reported by the datanode only when attached.
+      // Process the version report only after passing the fence check. Calling processVersionReport
+      // for fenced nodes would invoke shouldFenceDatanode a second time via sendFinalizeToDatanodeIfNeeded,
+      // producing duplicate log entries. Version details for fenced nodes are captured above.
       if (heartbeat.hasDatanodeVersion()) {
         LOG.debug("Processing DataNode version report.");
         nodeManager.processVersionReport(datanodeDetails, heartbeat.getDatanodeVersion());
@@ -104,6 +118,7 @@ public List<SCMCommand<?>> dispatch(SCMHeartbeatRequestProto heartbeat) {
       if (heartbeat.hasCommandQueueReport()) {
         commandQueueReport = heartbeat.getCommandQueueReport();
       }
+
       // should we dispatch heartbeat through eventPublisher?
       commands = nodeManager.processHeartbeat(datanodeDetails, commandQueueReport);
       if (heartbeat.hasNodeReport()) {
@@ -182,6 +197,24 @@ public List<SCMCommand<?>> dispatch(SCMHeartbeatRequestProto heartbeat) {
     return commands;
   }
 
+  /**
+   * Returns true if the node is not healthy (stale or dead) and its reported version is
+   * incompatible with this SCM. Healthy nodes are always allowed through — their version was
+   * already validated at registration. If the node status cannot be determined, returns false
+   * so the heartbeat is processed normally.
+   */
+  private boolean isUnhealthyWithIncompatibleVersion(DatanodeDetails datanodeDetails,
+      SCMHeartbeatRequestProto heartbeat) {
+    try {
+      return !nodeManager.getNodeStatus(datanodeDetails).isHealthy()
+          && nodeManager.shouldFenceDatanode(datanodeDetails, heartbeat.getDatanodeVersion());
+    } catch (NodeNotFoundException e) {
+      LOG.warn("Node {} not found when checking for version compatibility. " +
+          "This should not happen since the node is registered.", datanodeDetails, e);
+      return false;
+    }
+  }
+
   /**
    * Wrapper class for events with the datanode origin.
    */

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/TestSCMDatanodeHeartbeatDispatcher.java

@@ -25,12 +25,14 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import java.io.IOException;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.DatanodeID;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandStatusReportsProto;
@@ -39,21 +41,25 @@
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeReportProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMHeartbeatRequestProto;
 import org.apache.hadoop.hdds.scm.node.NodeManager;
+import org.apache.hadoop.hdds.scm.node.NodeStatus;
+import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException;
 import org.apache.hadoop.hdds.scm.server.SCMDatanodeHeartbeatDispatcher.CommandStatusReportFromDatanode;
 import org.apache.hadoop.hdds.scm.server.SCMDatanodeHeartbeatDispatcher.ContainerReportFromDatanode;
 import org.apache.hadoop.hdds.scm.server.SCMDatanodeHeartbeatDispatcher.NodeReportFromDatanode;
 import org.apache.hadoop.hdds.server.events.Event;
 import org.apache.hadoop.hdds.server.events.EventPublisher;
 import org.apache.hadoop.ozone.protocol.commands.ReregisterCommand;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
 
 /**
  * This class tests the behavior of SCMDatanodeHeartbeatDispatcher.
  */
 public class TestSCMDatanodeHeartbeatDispatcher {
 
   @Test
-  public void testNodeReportDispatcher() throws IOException {
+  public void testNodeReportDispatcher() throws IOException, NodeNotFoundException {
 
     AtomicInteger eventReceived = new AtomicInteger();
 
@@ -62,6 +68,7 @@ public void testNodeReportDispatcher() throws IOException {
     NodeManager mockNodeManager = mock(NodeManager.class);
     when(mockNodeManager.isNodeRegistered(any()))
         .thenReturn(true);
+    when(mockNodeManager.getNodeStatus(any())).thenReturn(NodeStatus.inServiceHealthy());
 
     SCMDatanodeHeartbeatDispatcher dispatcher =
         new SCMDatanodeHeartbeatDispatcher(mockNodeManager,
@@ -89,7 +96,7 @@ public <PAYLOAD, EVENT extends Event<PAYLOAD>> void fireEvent(
   }
 
   @Test
-  public void testContainerReportDispatcher() throws IOException {
+  public void testContainerReportDispatcher() throws IOException, NodeNotFoundException {
 
 
     AtomicInteger eventReceived = new AtomicInteger();
@@ -102,6 +109,7 @@ public void testContainerReportDispatcher() throws IOException {
     NodeManager mockNodeManager = mock(NodeManager.class);
     when(mockNodeManager.isNodeRegistered(any()))
         .thenReturn(true);
+    when(mockNodeManager.getNodeStatus(any())).thenReturn(NodeStatus.inServiceHealthy());
 
     SCMDatanodeHeartbeatDispatcher dispatcher =
         new SCMDatanodeHeartbeatDispatcher(
@@ -141,9 +149,10 @@ public <PAYLOAD, EVENT extends Event<PAYLOAD>> void fireEvent(
   }
 
   @Test
-  public void testVersionReportProcessedOnlyWhenPresent() throws IOException {
+  public void testVersionReportProcessedOnlyWhenPresent() throws IOException, NodeNotFoundException {
     NodeManager mockNodeManager = mock(NodeManager.class);
     when(mockNodeManager.isNodeRegistered(any())).thenReturn(true);
+    when(mockNodeManager.getNodeStatus(any())).thenReturn(NodeStatus.inServiceHealthy());
 
     SCMDatanodeHeartbeatDispatcher dispatcher =
         new SCMDatanodeHeartbeatDispatcher(mockNodeManager, mock(EventPublisher.class));
@@ -166,6 +175,84 @@ public void testVersionReportProcessedOnlyWhenPresent() throws IOException {
     verify(mockNodeManager, times(1)).processVersionReport(any(), any());
   }
 
+  private static Stream<NodeStatus> unhealthyNodeStatuses() {
+    return Stream.of(NodeStatus.inServiceStale(), NodeStatus.inServiceDead());
+  }
+
+  @Test
+  public void testHealthyNodeWithVersionReportIsNeverFenced() throws Exception {
+    // Healthy nodes passed the version gate at registration; their version only changes on restart,
+    // which causes re-registration on its own. The dispatcher must never fence a healthy node.
+    NodeManager mockNodeManager = mock(NodeManager.class);
+    when(mockNodeManager.isNodeRegistered(any())).thenReturn(true);
+    when(mockNodeManager.getNodeStatus(any())).thenReturn(NodeStatus.inServiceHealthy());
+
+    SCMDatanodeHeartbeatDispatcher dispatcher =
+        new SCMDatanodeHeartbeatDispatcher(mockNodeManager, mock(EventPublisher.class));
+
+    DatanodeVersionProto version = DatanodeVersionProto.newBuilder()
+        .setApparentVersion(1).setSoftwareVersion(1).build();
+    SCMHeartbeatRequestProto heartbeat = SCMHeartbeatRequestProto.newBuilder()
+        .setDatanodeDetails(randomDatanodeDetails().getProtoBufMessage())
+        .setDatanodeVersion(version)
+        .build();
+
+    dispatcher.dispatch(heartbeat);
+
+    verify(mockNodeManager, never()).addDatanodeCommand(any(DatanodeID.class), any(ReregisterCommand.class));
+    verify(mockNodeManager, times(1)).processHeartbeat(any(), any());
+  }
+
+  @ParameterizedTest
+  @MethodSource("unhealthyNodeStatuses")
+  public void testUnhealthyNodeWithIncompatibleVersionGetsReregisterCommand(
+      NodeStatus nodeStatus) throws Exception {
+    NodeManager mockNodeManager = mock(NodeManager.class);
+    when(mockNodeManager.isNodeRegistered(any())).thenReturn(true);
+    when(mockNodeManager.getNodeStatus(any())).thenReturn(nodeStatus);
+    when(mockNodeManager.shouldFenceDatanode(any(), any())).thenReturn(true);
+
+    SCMDatanodeHeartbeatDispatcher dispatcher =
+        new SCMDatanodeHeartbeatDispatcher(mockNodeManager, mock(EventPublisher.class));
+
+    DatanodeVersionProto version = DatanodeVersionProto.newBuilder()
+        .setApparentVersion(1).setSoftwareVersion(1).build();
+    SCMHeartbeatRequestProto heartbeat = SCMHeartbeatRequestProto.newBuilder()
+        .setDatanodeDetails(randomDatanodeDetails().getProtoBufMessage())
+        .setDatanodeVersion(version)
+        .build();
+
+    dispatcher.dispatch(heartbeat);
+
+    verify(mockNodeManager, times(1)).addDatanodeCommand(any(DatanodeID.class), any(ReregisterCommand.class));
+    verify(mockNodeManager, never()).processHeartbeat(any(), any());
+  }
+
+  @ParameterizedTest
+  @MethodSource("unhealthyNodeStatuses")
+  public void testUnhealthyNodeWithCompatibleVersionProcessesHeartbeatNormally(
+      NodeStatus nodeStatus) throws Exception {
+    NodeManager mockNodeManager = mock(NodeManager.class);
+    when(mockNodeManager.isNodeRegistered(any())).thenReturn(true);
+    when(mockNodeManager.getNodeStatus(any())).thenReturn(nodeStatus);
+    when(mockNodeManager.shouldFenceDatanode(any(), any())).thenReturn(false);
+
+    SCMDatanodeHeartbeatDispatcher dispatcher =
+        new SCMDatanodeHeartbeatDispatcher(mockNodeManager, mock(EventPublisher.class));
+
+    DatanodeVersionProto version = DatanodeVersionProto.newBuilder()
+        .setApparentVersion(1).setSoftwareVersion(1).build();
+    SCMHeartbeatRequestProto heartbeat = SCMHeartbeatRequestProto.newBuilder()
+        .setDatanodeDetails(randomDatanodeDetails().getProtoBufMessage())
+        .setDatanodeVersion(version)
+        .build();
+
+    dispatcher.dispatch(heartbeat);
+
+    verify(mockNodeManager, never()).addDatanodeCommand(any(DatanodeID.class), any(ReregisterCommand.class));
+    verify(mockNodeManager, times(1)).processHeartbeat(any(), any());
+  }
+
   /**
    * Asserts scm informs datanodes to re-register on a restart.
    *

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconNodeManager.java

@@ -306,7 +306,7 @@ protected void sendFinalizeToDatanodeIfNeeded(DatanodeDetails datanodeDetails,
    * it has finalized, unlike SCM.
    */
   @Override
-  protected boolean shouldFenceDatanode(DatanodeDetails dnDetails, DatanodeVersionProto versionReport) {
+  public boolean shouldFenceDatanode(DatanodeDetails dnDetails, DatanodeVersionProto versionReport) {
     if (versionReport == null || !versionReport.hasSoftwareVersion() || !versionReport.hasApparentVersion()) {
       LOG.error("Datanode {} did not report its version. Not allowing it to join the cluster.", dnDetails);
       return true;