HDFS-13048. LowRedundancyReplicatedBlocks metric can be negative

(cherry picked from commit 4aef8bd2efd68bf96c077ddda1538dcd5691b437)

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

@@ -365,7 +365,7 @@ class LowRedundancyBlocks implements Iterable<BlockInfo> {
           NameNode.blockStateChangeLog.debug(
               "BLOCK* NameSystem.LowRedundancyBlock.remove: Removing block" +
                   " {} from priority queue {}", block, i);
-          decrementBlockStat(block, priLevel, oldExpectedReplicas);
+          decrementBlockStat(block, i, oldExpectedReplicas);
           return true;
         }
       }

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

@@ -123,6 +123,21 @@ public class TestLowRedundancyBlockQueues {
     verifyBlockStats(queues, 2, 3, 2, 0, 0);
   }
 
+  @Test
+  public void testRemoveWithWrongPriority() {
+    final LowRedundancyBlocks queues = new LowRedundancyBlocks();
+    final BlockInfo corruptBlock = genBlockInfo(1);
+    assertAdded(queues, corruptBlock, 0, 0, 3);
+    assertInLevel(queues, corruptBlock,
+        LowRedundancyBlocks.QUEUE_WITH_CORRUPT_BLOCKS);
+    verifyBlockStats(queues, 0, 1, 0, 0, 0);
+
+    // Remove with wrong priority
+    queues.remove(corruptBlock, LowRedundancyBlocks.QUEUE_LOW_REDUNDANCY);
+    // Verify the number of corrupt block is decremented
+    verifyBlockStats(queues, 0, 0, 0, 0, 0);
+  }
+
   @Test
   public void testStripedBlockPriorities() throws Throwable {
     int dataBlkNum = ecPolicy.getNumDataUnits();