HDFS-15945. DataNodes with zero capacity and zero blocks should be decommissioned immediately.

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java

@@ -4584,8 +4584,14 @@ void processExtraRedundancyBlocksOnInService(
    */
   boolean isNodeHealthyForDecommissionOrMaintenance(DatanodeDescriptor node) {
     if (!node.checkBlockReportReceived()) {
-      LOG.info("Node {} hasn't sent its first block report.", node);
-      return false;
+      if (node.getNumBlocks() == 0) {
+        LOG.info("The number of blocks of {} are zero. "
+            + "Safe to decommission or put in maintenance.", node);
+        return true;
+      } else {
+        LOG.info("Node {} hasn't sent its first block report.", node);
+        return false;
+      }
     }
 
     if (node.isAlive()) {

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDecommission.java

@@ -1587,4 +1587,37 @@ public Boolean get() {
 
     cleanupFile(fileSys, file);
   }
+
+  /**
+   * DataNodes with 0 blocks should be decommissioned immediately
+   * even if they haven't reported the first block report.
+   */
+  @Test(timeout=60000)
+  public void testZeroBlocksNodesDecommission() throws Exception {
+    int numNamenodes = 1;
+    int numDatanodes = 3;
+    startCluster(numNamenodes, numDatanodes);
+
+    BlockManager bm = getCluster().getNamesystem().getBlockManager();
+    DatanodeManager dm = bm.getDatanodeManager();
+    DataNode dataNode = getCluster().getDataNodes().get(numDatanodes-1);
+    DatanodeID dnID = dataNode.getDatanodeId();
+    DatanodeDescriptor zeroBlocksNode = dm.getDatanode(dnID);
+    // clear the block report count
+    zeroBlocksNode.updateRegInfo(dnID);
+    // disable heartbeat not to send the first block report
+    DataNodeTestUtils.setHeartbeatsDisabledForTests(dataNode, true);
+
+    // decommission the datanode with 0 blocks
+    ArrayList<String> nodes = new ArrayList<>();
+    nodes.add(zeroBlocksNode.getXferAddr());
+    initExcludeHosts(nodes);
+    refreshNodes(0);
+    waitNodeState(zeroBlocksNode, AdminStates.DECOMMISSIONED);
+
+    // it should be decommissioned immediately
+    FSNamesystem ns = getCluster().getNamesystem(0);
+    int liveDecommissioned = ns.getNumDecomLiveDataNodes();
+    assertEquals(1, liveDecommissioned);
+  }
 }