HDFS-17093. Fix block report lease issue to avoid missing some storages report.

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

@@ -2866,14 +2866,18 @@ public boolean checkBlockReportLease(BlockReportContext context,
    * The given storage is reporting all its blocks.
    * Update the (storage{@literal -->}block list) and
    * (block{@literal -->}storage list) maps.
+   * totalReportNum -> totalStorageReportsNum
+   * currentReportNum -> currentStorageReportIndex
    *
    * @return true if all known storages of the given DN have finished reporting.
    * @throws IOException
    */
   public boolean processReport(final DatanodeID nodeID,
       final DatanodeStorage storage,
       final BlockListAsLongs newReport,
-      BlockReportContext context) throws IOException {
+      BlockReportContext context,
+      int totalReportNum,
+      int currentReportNum) throws IOException {
     namesystem.writeLock();
     final long startTime = Time.monotonicNow(); //after acquiring write lock
     final long endTime;
@@ -2904,7 +2908,8 @@ public boolean processReport(final DatanodeID nodeID,
       }
       if (namesystem.isInStartupSafeMode()
           && !StorageType.PROVIDED.equals(storageInfo.getStorageType())
-          && storageInfo.getBlockReportCount() > 0) {
+          && storageInfo.getBlockReportCount() > 0
+          && totalReportNum == currentReportNum) {
         blockLog.info("BLOCK* processReport 0x{} with lease ID 0x{}: "
             + "discarded non-initial block report from {}"
             + " because namenode still in startup phase",

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java

@@ -1650,7 +1650,7 @@ public DatanodeCommand blockReport(final DatanodeRegistration nodeReg,
           final int index = r;
           noStaleStorages = bm.runBlockOp(() ->
             bm.processReport(nodeReg, reports[index].getStorage(),
-                blocks, context));
+                blocks, context, reports.length, index + 1));
         }
       } else {
         throw new InvalidBlockReportLeaseException(context.getReportId(), context.getLeaseId());

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestBlockManager.java

@@ -1080,12 +1080,12 @@ public void testSafeModeIBR() throws Exception {
     reset(node);
 
     bm.processReport(node, new DatanodeStorage(ds.getStorageID()),
-        BlockListAsLongs.EMPTY, null);
+        BlockListAsLongs.EMPTY, null, 1, 1);
     assertEquals(1, ds.getBlockReportCount());
     // send block report again, should NOT be processed
     reset(node);
     bm.processReport(node, new DatanodeStorage(ds.getStorageID()),
-        BlockListAsLongs.EMPTY, null);
+        BlockListAsLongs.EMPTY, null, 1, 1);
     assertEquals(1, ds.getBlockReportCount());
 
     // re-register as if node restarted, should update existing node
@@ -1096,7 +1096,7 @@ public void testSafeModeIBR() throws Exception {
     // send block report, should be processed after restart
     reset(node);
     bm.processReport(node, new DatanodeStorage(ds.getStorageID()),
-                     BlockListAsLongs.EMPTY, null);
+                     BlockListAsLongs.EMPTY, null, 1, 1);
     // Reinitialize as registration with empty storage list pruned
     // node.storageMap.
     ds = node.getStorageInfos()[0];
@@ -1125,7 +1125,7 @@ public void testSafeModeIBRAfterIncremental() throws Exception {
     reset(node);
     doReturn(1).when(node).numBlocks();
     bm.processReport(node, new DatanodeStorage(ds.getStorageID()),
-        BlockListAsLongs.EMPTY, null);
+        BlockListAsLongs.EMPTY, null, 1, 1);
     assertEquals(1, ds.getBlockReportCount());
   }
 
@@ -1198,7 +1198,7 @@ public void testSafeModeIBRBeforeFirstFullBR() throws Exception {
     // Make sure it's the first full report
     assertEquals(0, ds.getBlockReportCount());
     bm.processReport(node, new DatanodeStorage(ds.getStorageID()),
-        builder.build(), null);
+        builder.build(), null, 1, 1);
     assertEquals(1, ds.getBlockReportCount());
 
     // verify the storage info is correct
@@ -1249,12 +1249,12 @@ public void testSafeModeWithProvidedStorageBR() throws Exception {
     bmPs.getDatanodeManager().addDatanode(node1);
 
     // process reports of provided storage and disk storage
-    bmPs.processReport(node0, providedStorage, BlockListAsLongs.EMPTY, null);
+    bmPs.processReport(node0, providedStorage, BlockListAsLongs.EMPTY, null, 2, 1);
     bmPs.processReport(node0, new DatanodeStorage(ds0.getStorageID()),
-        BlockListAsLongs.EMPTY, null);
-    bmPs.processReport(node1, providedStorage, BlockListAsLongs.EMPTY, null);
+        BlockListAsLongs.EMPTY, null, 2, 2);
+    bmPs.processReport(node1, providedStorage, BlockListAsLongs.EMPTY, null, 2, 1);
     bmPs.processReport(node1, new DatanodeStorage(ds1.getStorageID()),
-        BlockListAsLongs.EMPTY, null);
+        BlockListAsLongs.EMPTY, null, 2, 2);
 
     // The provided stoage report should not affect disk storage report
     DatanodeStorageInfo dsPs =

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestBlockReportLease.java

@@ -23,6 +23,8 @@
 import org.apache.hadoop.hdfs.protocol.BlockListAsLongs;
 import org.apache.hadoop.hdfs.server.datanode.DataNode;
 import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
+import org.apache.hadoop.hdfs.server.namenode.NameNode;
+import org.apache.hadoop.hdfs.server.namenode.NameNodeAdapter;
 import org.apache.hadoop.hdfs.server.protocol.BlockReportContext;
 import org.apache.hadoop.hdfs.server.protocol.DatanodeCommand;
 import org.apache.hadoop.hdfs.server.protocol.DatanodeRegistration;
@@ -53,6 +55,7 @@
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.doReturn;
 
 /**
  * Tests that BlockReportLease in BlockManager.
@@ -269,4 +272,88 @@ private StorageBlockReport[] createReports(DatanodeStorage[] dnStorages,
     }
     return storageBlockReports;
   }
+
+  @Test
+  public void testFirstIncompleteBlockReport() throws Exception {
+    HdfsConfiguration conf = new HdfsConfiguration();
+    Random rand = new Random();
+
+    try (MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf)
+            .numDataNodes(1).build()) {
+      cluster.waitActive();
+
+      FSNamesystem fsn = cluster.getNamesystem();
+
+      NameNode nameNode = cluster.getNameNode();
+      // pretend to be in safemode
+      NameNodeAdapter.enterSafeMode(nameNode, false);
+
+      BlockManager blockManager = fsn.getBlockManager();
+      BlockManager spyBlockManager = spy(blockManager);
+      fsn.setBlockManagerForTesting(spyBlockManager);
+      String poolId = cluster.getNamesystem().getBlockPoolId();
+
+      NamenodeProtocols rpcServer = cluster.getNameNodeRpc();
+
+      // Test based on one DataNode report to Namenode
+      DataNode dn = cluster.getDataNodes().get(0);
+      DatanodeDescriptor datanodeDescriptor = spyBlockManager
+              .getDatanodeManager().getDatanode(dn.getDatanodeId());
+
+      DatanodeRegistration dnRegistration = dn.getDNRegistrationForBP(poolId);
+      StorageReport[] storages = dn.getFSDataset().getStorageReports(poolId);
+
+      // Send heartbeat and request full block report lease
+      HeartbeatResponse hbResponse = rpcServer.sendHeartbeat(
+              dnRegistration, storages, 0, 0, 0, 0, 0, null, true,
+              SlowPeerReports.EMPTY_REPORT, SlowDiskReports.EMPTY_REPORT);
+
+      DelayAnswer delayer = new DelayAnswer(BlockManager.LOG);
+      doAnswer(delayer).when(spyBlockManager).processReport(
+              any(DatanodeStorageInfo.class),
+              any(BlockListAsLongs.class));
+
+      ExecutorService pool = Executors.newFixedThreadPool(1);
+
+      // Trigger sendBlockReport
+      BlockReportContext brContext = new BlockReportContext(1, 0,
+              rand.nextLong(), hbResponse.getFullBlockReportLeaseId());
+      // Build every storage with 100 blocks for sending report
+      DatanodeStorage[] datanodeStorages
+              = new DatanodeStorage[storages.length];
+      for (int i = 0; i < storages.length; i++) {
+        datanodeStorages[i] = storages[i].getStorage();
+        StorageBlockReport[] reports = createReports(datanodeStorages, 100);
+
+        // The first multiple send once, simulating the failure of the first report, only send successfully once
+        if(i == 0){
+          rpcServer.blockReport(dnRegistration, poolId, reports, brContext);
+        }
+
+        // Send blockReport
+        DatanodeCommand datanodeCommand = rpcServer.blockReport(dnRegistration, poolId, reports,
+                brContext);
+
+        // Wait until BlockManager calls processReport
+        delayer.waitForCall();
+
+        // Remove full block report lease about dn
+        spyBlockManager.getBlockReportLeaseManager()
+                .removeLease(datanodeDescriptor);
+
+        // Allow blockreport to proceed
+        delayer.proceed();
+
+        // Get result, it will not null if process successfully
+        assertTrue(datanodeCommand instanceof FinalizeCommand);
+        assertEquals(poolId, ((FinalizeCommand)datanodeCommand)
+                .getBlockPoolId());
+        if(i == 0){
+          assertEquals(2, datanodeDescriptor.getStorageInfos()[i].getBlockReportCount());
+        }else{
+          assertEquals(1, datanodeDescriptor.getStorageInfos()[i].getBlockReportCount());
+        }
+      }
+    }
+  }
 }