HDFS-14852. Removing from LowRedundancyBlocks does not remove the block from all queues. Contributed by Fei Hui.

(cherry picked from commit 6a49bf9bffbe9bad1c719fc3a3b734f0060df70a)

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

@@ -2475,7 +2475,7 @@ public class BlockManager implements BlockStatsMXBean {
            * with the most up-to-date block information (e.g. genstamp).
            */
           BlockInfo bi = blocksMap.getStoredBlock(timedOutItems[i]);
-          if (bi == null) {
+          if (bi == null || bi.isDeleted()) {
             continue;
           }
           NumberReplicas num = countNodes(timedOutItems[i]);

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

@@ -382,17 +382,18 @@ class LowRedundancyBlocks implements Iterable<BlockInfo> {
     } else {
       // Try to remove the block from all queues if the block was
       // not found in the queue for the given priority level.
+      boolean found = false;
       for (int i = 0; i < LEVEL; i++) {
         if (i != priLevel && priorityQueues.get(i).remove(block)) {
           NameNode.blockStateChangeLog.debug(
               "BLOCK* NameSystem.LowRedundancyBlock.remove: Removing block" +
                   " {} from priority queue {}", block, i);
           decrementBlockStat(block, i, oldExpectedReplicas);
-          return true;
+          found = true;
         }
       }
+      return found;
     }
-    return false;
   }
 
   private void decrementBlockStat(BlockInfo blockInfo, int priLevel,

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestLowRedundancyBlockQueues.java

@@ -276,4 +276,15 @@ public class TestLowRedundancyBlockQueues {
     }
     fail("Block " + block + " not found in level " + level);
   }
+
+  @Test
+  public void testRemoveBlockInManyQueues() {
+    LowRedundancyBlocks neededReconstruction = new LowRedundancyBlocks();
+    BlockInfo block = new BlockInfoContiguous(new Block(), (short)1024);
+    neededReconstruction.add(block, 2, 0, 1, 3);
+    neededReconstruction.add(block, 0, 0, 0, 3);
+    neededReconstruction.remove(block, LowRedundancyBlocks.LEVEL);
+    assertFalse("Should not contain the block.",
+        neededReconstruction.contains(block));
+  }
 }