diff --git a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
index 6390499d388..7a5cd02761c 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
+++ b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
@@ -628,6 +628,9 @@ Release 2.1.0-beta - UNRELEASED
     HDFS-4661. A few little code cleanups of some HDFS-347-related code. (Colin
     Patrick McCabe via atm)
 
+    HDFS-4878. On Remove Block, block is not removed from neededReplications
+    queue. (Tao Luo via shv)
+
   BREAKDOWN OF HADOOP-8562 and HDFS-3602 SUBTASKS AND RELATED JIRAS
 
     HDFS-4145. Merge hdfs cmd line scripts from branch-1-win. (David Lao,
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java
index 26c127bf965..7e8ef80cfad 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java
@@ -2869,8 +2869,9 @@ public void removeBlock(Block block) {
     addToInvalidates(block);
     corruptReplicas.removeFromCorruptReplicasMap(block);
     blocksMap.removeBlock(block);
-    // Remove the block from pendingReplications
+    // Remove the block from pendingReplications and neededReplications
     pendingReplications.remove(block);
+    neededReplications.remove(block, UnderReplicatedBlocks.LEVEL);
     if (postponedMisreplicatedBlocks.remove(block)) {
       postponedMisreplicatedBlocksCount.decrementAndGet();
     }
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestMetaSave.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestMetaSave.java
index c0775a62504..1681bee8403 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestMetaSave.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestMetaSave.java
@@ -87,17 +87,72 @@ public void testMetaSave() throws IOException, InterruptedException {
         + "metasave.out.txt";
     FileInputStream fstream = new FileInputStream(logFile);
     DataInputStream in = new DataInputStream(fstream);
-    BufferedReader reader = new BufferedReader(new InputStreamReader(in));
-    String line = reader.readLine();
-    assertTrue(line.equals(
-      "3 files and directories, 2 blocks = 5 total filesystem objects"));
-    line = reader.readLine();
-    assertTrue(line.equals("Live Datanodes: 1"));
-    line = reader.readLine();
-    assertTrue(line.equals("Dead Datanodes: 1"));
-    line = reader.readLine();
-    line = reader.readLine();
-    assertTrue(line.matches("^/filestatus[01]:.*"));
+    BufferedReader reader = null;
+    try {
+      reader = new BufferedReader(new InputStreamReader(in));
+      String line = reader.readLine();
+      assertTrue(line.equals(
+          "3 files and directories, 2 blocks = 5 total filesystem objects"));
+      line = reader.readLine();
+      assertTrue(line.equals("Live Datanodes: 1"));
+      line = reader.readLine();
+      assertTrue(line.equals("Dead Datanodes: 1"));
+      line = reader.readLine();
+      line = reader.readLine();
+      assertTrue(line.matches("^/filestatus[01]:.*"));
+    } finally {
+      if (reader != null)
+        reader.close();
+    }
+  }
+
+  /**
+   * Tests metasave after delete, to make sure there are no orphaned blocks
+   */
+  @Test
+  public void testMetasaveAfterDelete()
+      throws IOException, InterruptedException {
+
+    final FSNamesystem namesystem = cluster.getNamesystem();
+
+    for (int i = 0; i < 2; i++) {
+      Path file = new Path("/filestatus" + i);
+      DFSTestUtil.createFile(fileSys, file, 1024, 1024, blockSize, (short) 2,
+          seed);
+    }
+
+    cluster.stopDataNode(1);
+    // wait for namenode to discover that a datanode is dead
+    Thread.sleep(15000);
+    namesystem.setReplication("/filestatus0", (short) 4);
+    namesystem.delete("/filestatus0", true);
+    namesystem.delete("/filestatus1", true);
+
+    namesystem.metaSave("metasaveAfterDelete.out.txt");
+
+    // Verification
+    String logFile = System.getProperty("hadoop.log.dir") + "/"
+        + "metasaveAfterDelete.out.txt";
+    BufferedReader reader = null;
+    try {
+      FileInputStream fstream = new FileInputStream(logFile);
+      DataInputStream in = new DataInputStream(fstream);
+      reader = new BufferedReader(new InputStreamReader(in));
+      reader.readLine();
+      String line = reader.readLine();
+      assertTrue(line.equals("Live Datanodes: 1"));
+      line = reader.readLine();
+      assertTrue(line.equals("Dead Datanodes: 1"));
+      line = reader.readLine();
+      assertTrue(line.equals("Metasave: Blocks waiting for replication: 0"));
+      line = reader.readLine();
+      assertTrue(line.equals("Mis-replicated blocks that have been postponed:"));
+      line = reader.readLine();
+      assertTrue(line.equals("Metasave: Blocks being replicated: 0"));
+    } finally {
+      if (reader != null)
+        reader.close();
+    }
   }
 
   @AfterClass