Add Unit Test and fix removal condition

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

@@ -3493,7 +3493,8 @@ private void queueReportedBlock(DatanodeStorageInfo storageInfo, Block block,
               " from datanode {} for later processing because {}.",
           block, reportedState, storageInfo.getDatanodeDescriptor(), reason);
     }
-    pendingDNMessages.enqueueReportedBlock(storageInfo, block, reportedState);
+    pendingDNMessages.enqueueReportedBlock(storageInfo, block, reportedState,
+        isGenStampInFuture(block));
   }
 
   /**

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

@@ -94,23 +94,26 @@ void removeAllMessagesForDatanode(DatanodeDescriptor dn) {
   }
 
   void enqueueReportedBlock(DatanodeStorageInfo storageInfo, Block block,
-      ReplicaState reportedState) {
+      ReplicaState reportedState, boolean isGenStampInFuture) {
     long genStamp = block.getGenerationStamp();
+    Queue<ReportedBlockInfo> queue = null;
     if (BlockIdManager.isStripedBlockID(block.getBlockId())) {
       Block blkId = new Block(BlockIdManager.convertToStripedID(block
           .getBlockId()));
-      Queue<ReportedBlockInfo> queue = getBlockQueue(blkId);
-      queue.removeIf(rbi -> rbi.storageInfo.equals(storageInfo) &&
-          rbi.block.getGenerationStamp() < genStamp);
-      queue.add(new ReportedBlockInfo(storageInfo, new Block(block), reportedState));
+      queue = getBlockQueue(blkId);
     } else {
       block = new Block(block);
-      Queue<ReportedBlockInfo> queue = getBlockQueue(block);
-      // Remove the existing reports for this block since they are probably older and out of date
+      queue = getBlockQueue(block);
+    }
+    // We only want the latest non-future reported block to be queued for each
+    // DataNode. Otherwise, there can be a race condition that causes an old
+    // reported block to be kept in the queue until the SNN switches to ANN and
+    // the old reported block will be processed and marked as corrupt by the ANN.
+    if (!isGenStampInFuture) {
       queue.removeIf(rbi -> rbi.storageInfo.equals(storageInfo) &&
           rbi.block.getGenerationStamp() < genStamp);
-      queue.add(new ReportedBlockInfo(storageInfo, block, reportedState));
     }
+    queue.add(new ReportedBlockInfo(storageInfo, new Block(block), reportedState));
     count++;
   }
 

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

@@ -55,8 +55,8 @@ public void testQueues() {
     DatanodeDescriptor fakeDN = DFSTestUtil.getLocalDatanodeDescriptor();
     DatanodeStorage storage = new DatanodeStorage("STORAGE_ID");
     DatanodeStorageInfo storageInfo = new DatanodeStorageInfo(fakeDN, storage);
-    msgs.enqueueReportedBlock(storageInfo, block1Gs1, ReplicaState.FINALIZED);
-    msgs.enqueueReportedBlock(storageInfo, block1Gs2, ReplicaState.FINALIZED);
+    msgs.enqueueReportedBlock(storageInfo, block1Gs1, ReplicaState.FINALIZED, true);
+    msgs.enqueueReportedBlock(storageInfo, block1Gs2, ReplicaState.FINALIZED, true);
 
     assertEquals(2, msgs.count());
 
@@ -77,6 +77,36 @@ public void testQueues() {
     assertEquals(0, msgs.count());
   }
 
+  @Test
+  public void testQueuesKeepLatestNonFutureOnly() {
+    DatanodeDescriptor fakeDN = DFSTestUtil.getLocalDatanodeDescriptor();
+    DatanodeStorage storage = new DatanodeStorage("STORAGE_ID");
+    DatanodeStorageInfo storageInfo = new DatanodeStorageInfo(fakeDN, storage);
+    msgs.enqueueReportedBlock(storageInfo, block1Gs1, ReplicaState.FINALIZED,
+        false);
+    // block1Gs1 should be removed from the queue since it is older than
+    // block1Gs2 and block1Gs2 is not in the future.
+    msgs.enqueueReportedBlock(storageInfo, block1Gs2, ReplicaState.FINALIZED,
+        false);
+
+    assertEquals(1, msgs.count());
+
+    // Nothing queued yet for block 2
+    assertNull(msgs.takeBlockQueue(block2Gs1));
+    assertEquals(1, msgs.count());
+
+    Queue<ReportedBlockInfo> q =
+        msgs.takeBlockQueue(block1Gs2DifferentInstance);
+    assertEquals(
+            "ReportedBlockInfo [block=blk_1_2, dn=/default-rack/127.0.0.1:9866, reportedState=FINALIZED]",
+        Joiner.on(",").join(q));
+    assertEquals(0, msgs.count());
+
+    // Should be null if we pull again
+    assertNull(msgs.takeBlockQueue(block1Gs1));
+    assertEquals(0, msgs.count());
+  }
+
   @Test
   public void testPendingDataNodeMessagesWithEC() throws Exception {
     ErasureCodingPolicy ecPolicy = SystemErasureCodingPolicies