public void invalidateMissingBlock(String bpid, Block block, boolean checkFiles) {

    // The replica seems is on its volume map but not on disk.
    // We can't confirm here is block file lost or disk failed.
    // If block lost:
    //    deleted local block file is completely unnecessary
    // If disk failed:
    //    deleted local block file here may lead to missing-block
    //    when it with only 1 replication left now.
    // So remove if from volume map notify namenode is ok.
    try (AutoCloseableLock lock = lockManager.writeLock(LockLevel.BLOCK_POOl,
        bpid)) {
      // Check if this block is on the volume map.
      ReplicaInfo replica = volumeMap.get(bpid, block);
      // Double-check block or meta file existence when checkFiles as true.
      if (replica != null && (!checkFiles ||
          (!replica.blockDataExists() || !replica.metadataExists()))) {
        volumeMap.remove(bpid, block);
        invalidate(bpid, replica);
      }
    }
  }
@Test
  public void testInvalidateMissingBlock() throws Exception {
    long blockSize = 1024;
    int heartbeatInterval = 1;
    HdfsConfiguration c = new HdfsConfiguration();
    c.setInt(DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY, heartbeatInterval);
    c.setLong(DFS_BLOCK_SIZE_KEY, blockSize);
    MiniDFSCluster cluster = new MiniDFSCluster.Builder(c).
        numDataNodes(1).build();
    try {
      cluster.waitActive();
      DFSTestUtil.createFile(cluster.getFileSystem(), new Path("/a"),
          blockSize, (short)1, 0);

      String bpid = cluster.getNameNode().getNamesystem().getBlockPoolId();
      DataNode dn = cluster.getDataNodes().get(0);
      FsDatasetImpl fsdataset = (FsDatasetImpl) dn.getFSDataset();
      List<ReplicaInfo> replicaInfos = fsdataset.getFinalizedBlocks(bpid);
      assertEquals(1, replicaInfos.size());

      ReplicaInfo replicaInfo = replicaInfos.get(0);
      String blockPath = replicaInfo.getBlockURI().getPath();
      String metaPath = replicaInfo.getMetadataURI().getPath();
      File blockFile = new File(blockPath);
      File metaFile = new File(metaPath);

      // Mock local block file not found when disk with some exception.
      fsdataset.invalidateMissingBlock(bpid, replicaInfo, false);

      // Assert local block file wouldn't be deleted from disk.
      assertTrue(blockFile.exists());
      // Assert block info would be removed from ReplicaMap.
      assertEquals("null",
          fsdataset.getReplicaString(bpid, replicaInfo.getBlockId()));
      BlockManager blockManager = cluster.getNameNode().
          getNamesystem().getBlockManager();
      GenericTestUtils.waitFor(() ->
          blockManager.getLowRedundancyBlocksCount() == 1, 100, 5000);

      // Mock local block file found when disk back to normal.
      FsVolumeSpi.ScanInfo info = new FsVolumeSpi.ScanInfo(
          replicaInfo.getBlockId(), blockFile.getParentFile().getAbsoluteFile(),
          blockFile.getName(), metaFile.getName(), replicaInfo.getVolume());
      fsdataset.checkAndUpdate(bpid, info);
      GenericTestUtils.waitFor(() ->
          blockManager.getLowRedundancyBlocksCount() == 0, 100, 5000);
    } finally {
      cluster.shutdown();
    }
  }