diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
index b9f8e07f67a5f..a7fa12fdb1ecf 100755
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
@@ -523,6 +523,12 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
   public static final int DFS_NAMENODE_BLOCK_DELETION_UNLOCK_INTERVAL_MS_DEFAULT =
       10;
 
+  public static final String DFS_NAMENODE_BLOCK_DELETION_INCREMENT_KEY =
+          "dfs.namenode.block.deletion.increment";
+  public static final int DFS_NAMENODE_BLOCK_DELETION_INCREMENT_DEFAULT = 1000;
+  public static final String DFS_NAMENODE_BLOCK_DELETION_ASYNC_KEY = "dfs.namenode.block.deletion.async";
+  public static final boolean DFS_NAMENODE_BLOCK_DELETION_ASYNC_DEFAULT = true;
+
   public static final String DFS_NAMENODE_SNAPSHOT_CAPTURE_OPENFILES =
       HdfsClientConfigKeys.DFS_NAMENODE_SNAPSHOT_CAPTURE_OPENFILES;
   public static final boolean DFS_NAMENODE_SNAPSHOT_CAPTURE_OPENFILES_DEFAULT =
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
index 2cb29dfef8e65..697e95f565d3c 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
@@ -508,6 +508,8 @@ private boolean isFromProxyUser(CallerContext ctx) {
   private final boolean standbyShouldCheckpoint;
   private final boolean isSnapshotTrashRootEnabled;
   private final int snapshotDiffReportLimit;
+  private final boolean blockDeletionAsync;
+  private final int blockDeletionIncrement;
 
   /**
    * Whether enable checkOperation when call getBlocks.
@@ -1067,6 +1069,15 @@ static FSNamesystem loadFromDisk(Configuration conf) throws IOException {
       this.allowOwnerSetQuota = conf.getBoolean(
           DFSConfigKeys.DFS_PERMISSIONS_ALLOW_OWNER_SET_QUOTA_KEY,
           DFSConfigKeys.DFS_PERMISSIONS_ALLOW_OWNER_SET_QUOTA_DEFAULT);
+      this.blockDeletionAsync = conf.getBoolean(
+              DFSConfigKeys.DFS_NAMENODE_BLOCK_DELETION_ASYNC_KEY,
+              DFSConfigKeys.DFS_NAMENODE_BLOCK_DELETION_ASYNC_DEFAULT);
+      this.blockDeletionIncrement = conf.getInt(
+              DFSConfigKeys.DFS_NAMENODE_BLOCK_DELETION_INCREMENT_KEY,
+              DFSConfigKeys.DFS_NAMENODE_BLOCK_DELETION_INCREMENT_DEFAULT);
+      Preconditions.checkArgument(blockDeletionIncrement > 0,
+              DFSConfigKeys.DFS_NAMENODE_BLOCK_DELETION_INCREMENT_KEY +
+                      " must be a positive integer.");
       this.isGetBlocksCheckOperationEnabled = conf.getBoolean(
           DFSConfigKeys.DFS_NAMENODE_GETBLOCKS_CHECK_OPERATION_KEY,
           DFSConfigKeys.DFS_NAMENODE_GETBLOCKS_CHECK_OPERATION_DEFAULT);
@@ -2391,8 +2402,7 @@ boolean truncate(String src, long newLength, String clientName,
       }
       getEditLog().logSync();
       if (!toRemoveBlocks.getToDeleteList().isEmpty()) {
-        blockManager.addBLocksToMarkedDeleteQueue(
-            toRemoveBlocks.getToDeleteList());
+        removeBlocks(toRemoveBlocks.getToDeleteList());
       }
       logAuditEvent(true, operationName, src, null, status);
     } catch (AccessControlException e) {
@@ -2842,8 +2852,7 @@ private HdfsFileStatus startFileInt(String src,
       if (!skipSync) {
         getEditLog().logSync();
         if (toRemoveBlocks != null) {
-          blockManager.addBLocksToMarkedDeleteQueue(
-              toRemoveBlocks.getToDeleteList());
+          removeBlocks(toRemoveBlocks.getToDeleteList());
         }
       }
     }
@@ -3371,8 +3380,7 @@ void renameTo(final String src, final String dst,
     assert res != null;
     BlocksMapUpdateInfo collectedBlocks = res.collectedBlocks;
     if (!collectedBlocks.getToDeleteList().isEmpty()) {
-      blockManager.addBLocksToMarkedDeleteQueue(
-          collectedBlocks.getToDeleteList());
+      removeBlocks(collectedBlocks.getToDeleteList());
     }
 
     logAuditEvent(true, operationName + " (options=" +
@@ -3411,12 +3419,39 @@ boolean delete(String src, boolean recursive, boolean logRetryCache)
     getEditLog().logSync();
     logAuditEvent(ret, operationName, src);
     if (toRemovedBlocks != null) {
-      blockManager.addBLocksToMarkedDeleteQueue(
-          toRemovedBlocks.getToDeleteList());
+      removeBlocks(toRemovedBlocks.getToDeleteList());
     }
     return ret;
   }
 
+  /**
+   * If blockDeletionAsync enables, blocks will be deleted asynchronously.
+   * If not, incrementally remove the blocks from blockManager
+   * Writelock is dropped and reacquired every BLOCK_DELETION_INCREMENT to
+   * ensure that other waiters on the lock can get in. See HDFS-2938
+   *
+   * @param toDeleteList
+   *          a list of blocks that need to be removed from blocksMap
+   */
+  void removeBlocks(List<BlockInfo> toDeleteList) {
+    if (this.blockDeletionAsync) {
+      blockManager.addBLocksToMarkedDeleteQueue(toDeleteList);
+    } else {
+      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");
+        }
+      }
+    }
+  }
+
+
   FSPermissionChecker getPermissionChecker()
       throws AccessControlException {
     return dir.getPermissionChecker();
@@ -4665,8 +4700,7 @@ private void clearCorruptLazyPersistFiles()
                   INodesInPath.fromINode((INodeFile) bc), false);
           changed |= toRemoveBlocks != null;
           if (toRemoveBlocks != null) {
-            blockManager.addBLocksToMarkedDeleteQueue(
-                toRemoveBlocks.getToDeleteList());
+            removeBlocks(toRemoveBlocks.getToDeleteList());
           }
         }
       } finally {
@@ -7462,8 +7496,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.addBLocksToMarkedDeleteQueue(
-          blocksToBeDeleted.getToDeleteList());
+      removeBlocks(blocksToBeDeleted.getToDeleteList());
     }
     logAuditEvent(true, operationName, rootPath, null, null);
   }
@@ -7489,8 +7522,7 @@ public void gcDeletedSnapshot(String snapshotRoot, String snapshotName)
     } finally {
       writeUnlock(operationName, getLockReportInfoSupplier(rootPath));
     }
-    blockManager.addBLocksToMarkedDeleteQueue(
-        blocksToBeDeleted.getToDeleteList());
+    removeBlocks(blocksToBeDeleted.getToDeleteList());
   }
 
   /**
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
index 94c3ea0cc9b0c..1cfcfb7deaee6 100755
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
@@ -6292,6 +6292,24 @@
     </description>
   </property>
 
+  <property>
+    <name>dfs.namenode.block.deletion.async</name>
+    <value>true</value>
+    <description>
+      If false, disable block deleting asynchronously
+    </description>
+  </property>
+
+  <property>
+    <name>dfs.namenode.block.deletion.increment</name>
+    <value>1000</value>
+    <description>
+      The number of block deletion increment.
+      This setting will control the block increment deletion rate to
+      ensure that other waiters on the lock can get in.
+    </description>
+  </property>
+
   <property>
     <name>dfs.namenode.rpc-address.auxiliary-ports</name>
     <value></value>