add pendingDeleteBlocksCount metrics and remove removeBlocks func

Author: zhuxiangyi

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

@@ -4940,13 +4940,15 @@ public long getLastRedundancyMonitorTS() {
   private class MarkedDeleteBlockScrubber implements Runnable {
     private Iterator<BlockInfo> toDeleteIterator = null;
     private boolean isSleep;
+    private NameNodeMetrics metrics;
 
     private void remove(long time) {
       if (checkToDeleteIterator()) {
         namesystem.writeLock();
         try {
           while (toDeleteIterator.hasNext()) {
             removeBlock(toDeleteIterator.next());
+            metrics.decrPendingDeleteBlocksCount();
             if (Time.monotonicNow() - time > deleteBlockLockTimeMs) {
               isSleep = true;
               break;
@@ -4969,7 +4971,7 @@ public void run() {
           !Thread.currentThread().isInterrupted()) {
         if (!markedDeleteQueue.isEmpty() || checkToDeleteIterator()) {
           try {
-            NameNodeMetrics metrics = NameNode.getNameNodeMetrics();
+            metrics = NameNode.getNameNodeMetrics();
             metrics.setDeleteBlocksQueued(markedDeleteQueue.size());
             isSleep = false;
             long startTime = Time.monotonicNow();
@@ -5351,10 +5353,17 @@ public BlockIdManager getBlockIdManager() {
     return blockIdManager;
   }
 
+  @VisibleForTesting
   public ConcurrentLinkedQueue<List<BlockInfo>> getMarkedDeleteQueue() {
     return markedDeleteQueue;
   }
 
+  public void addBLocksToMarkedDeleteQueue(List<BlockInfo> blockInfos) {
+    markedDeleteQueue.add(blockInfos);
+    NameNode.getNameNodeMetrics().
+        incrPendingDeleteBlocksCount(blockInfos.size());
+  }
+
   public long nextGenerationStamp(boolean legacyBlock) throws IOException {
     return blockIdManager.nextGenerationStamp(legacyBlock);
   }

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

@@ -98,7 +98,6 @@
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_SNAPSHOT_DIFF_LISTING_LIMIT_DEFAULT;
 import static org.apache.hadoop.hdfs.DFSUtil.isParentEntry;
 
-import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.atomic.AtomicLong;
 
 import org.apache.commons.text.CaseUtils;
@@ -2374,7 +2373,7 @@ boolean truncate(String src, long newLength, String clientName,
       }
       getEditLog().logSync();
       if (!toRemoveBlocks.getToDeleteList().isEmpty()) {
-        blockManager.getMarkedDeleteQueue().add(
+        blockManager.addBLocksToMarkedDeleteQueue(
             toRemoveBlocks.getToDeleteList());
       }
       logAuditEvent(true, operationName, src, null, status);
@@ -2822,7 +2821,7 @@ private HdfsFileStatus startFileInt(String src,
       if (!skipSync) {
         getEditLog().logSync();
         if (toRemoveBlocks != null) {
-          blockManager.getMarkedDeleteQueue().add(
+          blockManager.addBLocksToMarkedDeleteQueue(
               toRemoveBlocks.getToDeleteList());
         }
       }
@@ -3346,7 +3345,7 @@ void renameTo(final String src, final String dst,
     assert res != null;
     BlocksMapUpdateInfo collectedBlocks = res.collectedBlocks;
     if (!collectedBlocks.getToDeleteList().isEmpty()) {
-      blockManager.getMarkedDeleteQueue().add(
+      blockManager.addBLocksToMarkedDeleteQueue(
           collectedBlocks.getToDeleteList());
     }
 
@@ -3386,7 +3385,7 @@ boolean delete(String src, boolean recursive, boolean logRetryCache)
     getEditLog().logSync();
     logAuditEvent(ret, operationName, src);
     if (toRemovedBlocks != null) {
-      blockManager.getMarkedDeleteQueue().add(
+      blockManager.addBLocksToMarkedDeleteQueue(
           toRemovedBlocks.getToDeleteList());
     }
     return ret;
@@ -3397,30 +3396,6 @@ FSPermissionChecker getPermissionChecker()
     return dir.getPermissionChecker();
   }
 
-  /**
-   * From the given list, incrementally remove the blocks from blockManager
-   * Writelock is dropped and reacquired every blockDeletionIncrement to
-   * ensure that other waiters on the lock can get in. See HDFS-2938
-   *
-   * @param blocks
-   *          An instance of {@link BlocksMapUpdateInfo} which contains a list
-   *          of blocks that need to be removed from blocksMap
-   */
-  void removeBlocks(BlocksMapUpdateInfo blocks) {
-    List<BlockInfo> toDeleteList = blocks.getToDeleteList();
-    Iterator<BlockInfo> iter = toDeleteList.iterator();
-    while (iter.hasNext()) {
-      writeLock();
-      try {
-        for (int i = 0; i < blockDeletionIncrement && iter.hasNext(); i++) {
-          blockManager.removeBlock(iter.next());
-        }
-      } finally {
-        writeUnlock("removeBlocks");
-      }
-    }
-  }
-
   /**
    * Remove leases and inodes related to a given path
    * @param removedUCFiles INodes whose leases need to be released
@@ -4629,7 +4604,7 @@ private void clearCorruptLazyPersistFiles()
                   INodesInPath.fromINode((INodeFile) bc), false);
           changed |= toRemoveBlocks != null;
           if (toRemoveBlocks != null) {
-            blockManager.getMarkedDeleteQueue().add(
+            blockManager.addBLocksToMarkedDeleteQueue(
                 toRemoveBlocks.getToDeleteList());
           }
         }
@@ -7341,7 +7316,7 @@ void deleteSnapshot(String snapshotRoot, String snapshotName,
     // Breaking the pattern as removing blocks have to happen outside of the
     // global lock
     if (blocksToBeDeleted != null) {
-      blockManager.getMarkedDeleteQueue().add(
+      blockManager.addBLocksToMarkedDeleteQueue(
           blocksToBeDeleted.getToDeleteList());
     }
     logAuditEvent(true, operationName, rootPath, null, null);
@@ -7368,7 +7343,7 @@ public void gcDeletedSnapshot(String snapshotRoot, String snapshotName)
     } finally {
       writeUnlock(operationName, getLockReportInfoSupplier(rootPath));
     }
-    blockManager.getMarkedDeleteQueue().add(
+    blockManager.addBLocksToMarkedDeleteQueue(
         blocksToBeDeleted.getToDeleteList());
   }
 

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

@@ -91,6 +91,8 @@ public class NameNodeMetrics {
   MutableGaugeInt pendingEditsCount;
   @Metric("Number of delete blocks Queued")
   MutableGaugeInt deleteBlocksQueued;
+  @Metric("Number of pending deletion blocks")
+  MutableGaugeInt pendingDeleteBlocksCount;
 
   @Metric("Number of file system operations")
   public long totalFileOps(){
@@ -347,6 +349,14 @@ public void setDeleteBlocksQueued(int size) {
     deleteBlocksQueued.set(size);
   }
 
+  public void incrPendingDeleteBlocksCount(int size) {
+    pendingDeleteBlocksCount.incr(size);
+  }
+
+  public void decrPendingDeleteBlocksCount() {
+    pendingDeleteBlocksCount.decr();
+  }
+
   public void addBlockOpsBatched(int count) {
     blockOpsBatched.incr(count);
   }