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 9e1333f95295b..8e405ac09b352 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
@@ -495,6 +495,14 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
   public static final int DFS_NAMENODE_BLOCK_DELETION_UNLOCK_INTERVAL_MS_DEFAULT =
       10;
 
+  /** Block deletion increment. */
+  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;
+  /** Block deletion asynchronous. */
+  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 13894b4fecf8c..95a8cca87d7b8 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
@@ -501,6 +501,8 @@ private boolean isClientPortInfoAbsent(CallerContext ctx){
   private final String supergroup;
   private final boolean standbyShouldCheckpoint;
   private final boolean isSnapshotTrashRootEnabled;
+  private final boolean blockDeletionAsync;
+  private final int blockDeletionIncrement;
   private final int snapshotDiffReportLimit;
 
   /**
@@ -1064,6 +1066,12 @@ 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);
       this.isGetBlocksCheckOperationEnabled = conf.getBoolean(
           DFSConfigKeys.DFS_NAMENODE_GETBLOCKS_CHECK_OPERATION_KEY,
           DFSConfigKeys.DFS_NAMENODE_GETBLOCKS_CHECK_OPERATION_DEFAULT);
@@ -2380,8 +2388,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) {
@@ -2828,8 +2835,7 @@ private HdfsFileStatus startFileInt(String src,
       if (!skipSync) {
         getEditLog().logSync();
         if (toRemoveBlocks != null) {
-          blockManager.addBLocksToMarkedDeleteQueue(
-              toRemoveBlocks.getToDeleteList());
+          removeBlocks(toRemoveBlocks.getToDeleteList());
         }
       }
     }
@@ -3352,8 +3358,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=" +
@@ -3392,8 +3397,7 @@ 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;
   }
@@ -3403,6 +3407,33 @@ FSPermissionChecker getPermissionChecker()
     return dir.getPermissionChecker();
   }
 
+  /**
+   * 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");
+        }
+      }
+    }
+  }
+
   /**
    * Remove leases and inodes related to a given path
    * @param removedUCFiles INodes whose leases need to be released
@@ -4609,8 +4640,7 @@ private void clearCorruptLazyPersistFiles()
                   INodesInPath.fromINode((INodeFile) bc), false);
           changed |= toRemoveBlocks != null;
           if (toRemoveBlocks != null) {
-            blockManager.addBLocksToMarkedDeleteQueue(
-                toRemoveBlocks.getToDeleteList());
+            removeBlocks(toRemoveBlocks.getToDeleteList());
           }
         }
       } finally {
@@ -7348,8 +7378,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);
   }
@@ -7375,8 +7404,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 33ffd07c8de2b..15a64db81973b 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
@@ -6163,6 +6163,14 @@
     </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.rpc-address.auxiliary-ports</name>
     <value></value>