HDFS-17905. Clean up pendingDNMessages on block removal (#8415)

Author: kokonguyen191

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

@@ -4982,6 +4982,8 @@ public void removeBlock(BlockInfo block) {
       DatanodeStorageInfo.decrementBlocksScheduled(remove.getTargets()
           .toArray(new DatanodeStorageInfo[remove.getTargets().size()]));
     }
+    // Remove all pending messages for this deleted block from the queue
+    pendingDNMessages.takeBlockQueue(block);
     neededReconstruction.remove(block, LowRedundancyBlocks.LEVEL);
     postponedMisreplicatedBlocks.remove(block);
   }

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/ha/TestPendingCorruptDnMessages.java

@@ -139,4 +139,56 @@ private static boolean wipeAndRestartDn(MiniDFSCluster cluster, int dnIndex)
     return cluster.restartDataNode(dnProps, true);
   }
 
+  @Test
+  @Timeout(value = 60)
+  public void testRemoveBlockCleansUpPendingDNMessages() throws Exception {
+    HdfsConfiguration conf = new HdfsConfiguration();
+    conf.setInt(DFSConfigKeys.DFS_HA_TAILEDITS_PERIOD_KEY, 1);
+    final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf)
+        .numDataNodes(1)
+        .nnTopology(MiniDFSNNTopology.simpleHATopology())
+        .build();
+
+    try {
+      cluster.transitionToActive(0);
+
+      FileSystem fs = HATestUtil.configureFailoverFs(cluster, conf);
+      OutputStream out = fs.create(filePath);
+      out.write("foo bar baz".getBytes());
+      out.close();
+
+      HATestUtil.waitForStandbyToCatchUp(cluster.getNameNode(0), cluster.getNameNode(1));
+
+      // Send genstamp to the future
+      ExtendedBlock block = DFSTestUtil.getFirstBlock(fs, filePath);
+      cluster.changeGenStampOfBlock(0, block, 4200);
+
+      // Run directory scanner to update Datanode's volumeMap
+      DataNodeTestUtils.runDirectoryScanner(cluster.getDataNodes().get(0));
+      // Stop the DN so the replica with the changed gen stamp will be reported
+      // when this DN starts up.
+      DataNodeProperties dnProps = cluster.stopDataNode(0);
+
+      // Restart the namenode so that when the DN comes up it will see an initial
+      // block report.
+      cluster.restartNameNode(1, false);
+      assertTrue(cluster.restartDataNode(dnProps, true));
+
+      // Wait until the standby NN queues up the future block in the pending DN
+      // message queue.
+      GenericTestUtils.waitFor((Supplier<Boolean>) () ->
+              cluster.getNamesystem(1).getBlockManager().getPendingDataNodeMessageCount() == 1, 1000,
+          30000);
+
+      // Delete the file on active, make standby tail the edit
+      fs.delete(filePath, false);
+      cluster.getNameNode(0).getRpcServer().rollEditLog();
+      cluster.getNameNode(1).getNamesystem().getEditLogTailer().doTailEdits();
+
+      // SB pendingDNMessage should be empty now, else there's a leak
+      assertEquals(0, cluster.getNamesystem(1).getBlockManager().getPendingDataNodeMessageCount());
+    } finally {
+      cluster.shutdown();
+    }
+  }
 }