HDFS-16931. Observer nn delete blocks asynchronously when tail OP_DELETE editlog

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

@@ -2757,14 +2757,16 @@ public class BlockManager implements BlockStatsMXBean {
    * Removes the blocks from blocksmap and updates the safemode blocks total.
    * @param blocks An instance of {@link BlocksMapUpdateInfo} which contains a
    *               list of blocks that need to be removed from blocksMap
+   * @param asnyc whether to delete a block asynchronously
    */
-  public void removeBlocksAndUpdateSafemodeTotal(BlocksMapUpdateInfo blocks) {
+  public void removeBlocksAndUpdateSafemodeTotal(BlocksMapUpdateInfo blocks,boolean deleteAsync) {
     assert namesystem.hasWriteLock();
     // In the case that we are a Standby tailing edits from the
     // active while in safe-mode, we need to track the total number
     // of blocks and safe blocks in the system.
     boolean trackBlockCounts = bmSafeMode.isSafeModeTrackingBlocks();
     int numRemovedComplete = 0, numRemovedSafe = 0;
+    deleteAsync = deleteAsync && !trackBlockCounts;
 
     for (BlockInfo b : blocks.getToDeleteList()) {
       if (trackBlockCounts) {
@@ -2775,8 +2777,15 @@ public class BlockManager implements BlockStatsMXBean {
           }
         }
       }
-      removeBlock(b);
+      if (!deleteAsync){
+        removeBlock(b);
+      }
     }
+    // delete block async
+    if (deleteAsync) {
+      this.getMarkedDeleteQueue().add(blocks.getToDeleteList());
+    }
+
     if (trackBlockCounts) {
       if (LOG.isDebugEnabled()) {
         LOG.debug("Adjusting safe-mode totals for deletion."

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

@@ -127,8 +127,9 @@ class FSDirDeleteOp {
    * @param fsd the FSDirectory instance
    * @param iip inodes of a path to be deleted
    * @param mtime the time the inode is removed
+   * @param asnyc whether to delete a block asynchronously
    */
-  static void deleteForEditLog(FSDirectory fsd, INodesInPath iip, long mtime)
+  static void deleteForEditLog(FSDirectory fsd, INodesInPath iip, long mtime, boolean asnyc)
       throws IOException {
     assert fsd.hasWriteLock();
     FSNamesystem fsn = fsd.getFSNamesystem();
@@ -148,7 +149,7 @@ class FSDirDeleteOp {
     if (filesRemoved) {
       fsn.removeSnapshottableDirs(snapshottableDirs);
       fsn.removeLeasesAndINodes(removedUCFiles, removedINodes, false);
-      fsn.getBlockManager().removeBlocksAndUpdateSafemodeTotal(collectedBlocks);
+      fsn.getBlockManager().removeBlocksAndUpdateSafemodeTotal(collectedBlocks,asnyc);
     }
   }
 

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

@@ -346,7 +346,7 @@ class FSDirRenameOp {
         collectedBlocks, options);
     if (!collectedBlocks.getToDeleteList().isEmpty()) {
       fsd.getFSNamesystem().getBlockManager()
-          .removeBlocksAndUpdateSafemodeTotal(collectedBlocks);
+          .removeBlocksAndUpdateSafemodeTotal(collectedBlocks,false);
     }
   }
 

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

@@ -198,7 +198,7 @@ final class FSDirTruncateOp {
     }
     assert onBlockBoundary == (truncateBlock == null) :
       "truncateBlock is null iff on block boundary: " + truncateBlock;
-    fsn.getBlockManager().removeBlocksAndUpdateSafemodeTotal(collectedBlocks);
+    fsn.getBlockManager().removeBlocksAndUpdateSafemodeTotal(collectedBlocks,false);
   }
 
   /**

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

@@ -411,7 +411,7 @@ public class FSEditLogLoader {
       INodeFile oldFile = INodeFile.valueOf(iip.getLastINode(), path, true);
       if (oldFile != null && addCloseOp.overwrite) {
         // This is OP_ADD with overwrite
-        FSDirDeleteOp.deleteForEditLog(fsDir, iip, addCloseOp.mtime);
+        FSDirDeleteOp.deleteForEditLog(fsDir, iip, addCloseOp.mtime,false);
         iip = INodesInPath.replace(iip, iip.length() - 1, null);
         oldFile = null;
       }
@@ -627,7 +627,7 @@ public class FSEditLogLoader {
       final String src = renameReservedPathsOnUpgrade(
           deleteOp.path, logVersion);
       final INodesInPath iip = fsDir.getINodesInPath(src, DirOp.WRITE_LINK);
-      FSDirDeleteOp.deleteForEditLog(fsDir, iip, deleteOp.timestamp);
+      FSDirDeleteOp.deleteForEditLog(fsDir, iip, deleteOp.timestamp,true);
 
       if (toAddRetryCache) {
         fsNamesys.addCacheEntry(deleteOp.rpcClientId, deleteOp.rpcCallId);
@@ -823,7 +823,7 @@ public class FSEditLogLoader {
           new INode.ReclaimContext(fsNamesys.dir.getBlockStoragePolicySuite(),
               collectedBlocks, removedINodes, null), deleteSnapshotOp.mtime);
       fsNamesys.getBlockManager().removeBlocksAndUpdateSafemodeTotal(
-          collectedBlocks);
+          collectedBlocks,false);
       collectedBlocks.clear();
       fsNamesys.dir.removeFromInodeMap(removedINodes);
       removedINodes.clear();