void removeQueuedBlock(DatanodeStorageInfo storageInfo, Block block) {
    if (storageInfo == null || block == null) {
      return;
    }
    Block blk = new Block(block);
    if (BlockIdManager.isStripedBlockID(block.getBlockId())) {
      blk = new Block(BlockIdManager.convertToStripedID(block
          .getBlockId()));
    }
    Queue<ReportedBlockInfo> queue = queueByBlockId.get(blk);
    if (queue == null) {
      return;
    }
    // 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.
    // See HDFS-17453
    int size = queue.size();
    if (queue.removeIf(rbi -> storageInfo.equals(rbi.storageInfo))) {
      count -= (size - queue.size());
    }
    // If the block message queue is now empty, we should remove the block
    // from the queue.
    if (queue.isEmpty()) {
      queueByBlockId.remove(blk);
    }
  }
@Test
  public void testRemoveQueuedBlock() {
    DatanodeDescriptor fakeDN1 = DFSTestUtil.getDatanodeDescriptor(
        "localhost", 8898, "/default-rack");
    DatanodeDescriptor fakeDN2 = DFSTestUtil.getDatanodeDescriptor(
        "localhost", 8899, "/default-rack");
    DatanodeStorage storage1 = new DatanodeStorage("STORAGE_ID_1");
    DatanodeStorage storage2 = new DatanodeStorage("STORAGE_ID_2");
    DatanodeStorageInfo storageInfo1 = new DatanodeStorageInfo(fakeDN1, storage1);
    DatanodeStorageInfo storageInfo2 = new DatanodeStorageInfo(fakeDN2, storage2);
    msgs.enqueueReportedBlock(storageInfo1, block1Gs1, ReplicaState.FINALIZED);
    msgs.enqueueReportedBlock(storageInfo2, block1Gs1, ReplicaState.FINALIZED);
    msgs.enqueueReportedBlock(storageInfo1, block1Gs2, ReplicaState.FINALIZED);
    msgs.enqueueReportedBlock(storageInfo2, block1Gs2, ReplicaState.FINALIZED);
    List<ReportedBlockInfo> rbis = Arrays.asList(
        new ReportedBlockInfo(storageInfo2, block1Gs1, ReplicaState.FINALIZED),
        new ReportedBlockInfo(storageInfo2, block1Gs2, ReplicaState.FINALIZED));

    assertEquals(4, msgs.count());

    // Nothing queued yet for block 2
    assertNull(msgs.takeBlockQueue(block2Gs1));
    assertEquals(4, msgs.count());

    msgs.removeQueuedBlock(storageInfo1, block1Gs1);
    Queue<ReportedBlockInfo> q =
        msgs.takeBlockQueue(block1Gs2DifferentInstance);
    assertEquals(Joiner.on(",").join(rbis),
        Joiner.on(",").join(q));
    assertEquals(0, msgs.count());

    // Should be null if we pull again;
    assertNull(msgs.takeBlockQueue(block1Gs2));
    assertEquals(0, msgs.count());
  }