diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/common/Storage.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/common/Storage.java
index ea10f011f9..83a82566f6 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/common/Storage.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/common/Storage.java
@@ -801,8 +801,7 @@ public abstract class Storage extends StorageInfo {
       case RECOVER_UPGRADE:   // mv previous.tmp -> current
         LOG.info("Recovering storage directory {} from previous upgrade",
             rootPath);
-        if (curDir.exists())
-          deleteDir(curDir);
+        deleteAsync(curDir);
         rename(getPreviousTmp(), curDir);
         return;
       case COMPLETE_ROLLBACK: // rm removed.tmp
@@ -818,21 +817,19 @@ public abstract class Storage extends StorageInfo {
       case COMPLETE_FINALIZE: // rm finalized.tmp
         LOG.info("Completing previous finalize for storage directory {}",
             rootPath);
-        deleteDir(getFinalizedTmp());
+        deleteAsync(getFinalizedTmp());
         return;
       case COMPLETE_CHECKPOINT: // mv lastcheckpoint.tmp -> previous.checkpoint
         LOG.info("Completing previous checkpoint for storage directory {}",
             rootPath);
         File prevCkptDir = getPreviousCheckpoint();
-        if (prevCkptDir.exists())
-          deleteDir(prevCkptDir);
+        deleteAsync(prevCkptDir);
         rename(getLastCheckpointTmp(), prevCkptDir);
         return;
       case RECOVER_CHECKPOINT:  // mv lastcheckpoint.tmp -> current
         LOG.info("Recovering storage directory {} from failed checkpoint",
             rootPath);
-        if (curDir.exists())
-          deleteDir(curDir);
+        deleteAsync(curDir);
         rename(getLastCheckpointTmp(), curDir);
         return;
       default:
@@ -840,7 +837,30 @@ public abstract class Storage extends StorageInfo {
             + " for storage directory: " + rootPath);
       }
     }
-    
+
+    /**
+     * Rename the curDir to curDir.tmp and delete the curDir.tmp parallely.
+     * @throws IOException
+     */
+    private void deleteAsync(File curDir) throws IOException {
+      if (curDir.exists()) {
+        File curTmp = new File(curDir.getParent(), curDir.getName() + ".tmp");
+        if (curTmp.exists()) {
+          deleteDir(curTmp);
+        }
+        rename(curDir, curTmp);
+        new Thread("Async Delete Current.tmp") {
+          public void run() {
+            try {
+              deleteDir(curTmp);
+            } catch (IOException e) {
+              LOG.warn("Deleting storage directory {} failed", curTmp);
+            }
+          }
+        }.start();
+      }
+    }
+
     /**
      * @return true if the storage directory should prompt the user prior
      * to formatting (i.e if the directory appears to contain some data)