void processTimedOutExcessBlocks() {
    if (excessRedundancyMap.size() == 0) {
      return;
    }
    namesystem.writeLock();
    long now = Time.monotonicNow();
    int processed = 0;
    try {
      Iterator<Map.Entry<String, LightWeightHashSet<Block>>> iter =
          excessRedundancyMap.getExcessRedundancyMap().entrySet().iterator();
      while (iter.hasNext() && processed < excessRedundancyTimeoutCheckLimit) {
        Map.Entry<String, LightWeightHashSet<Block>> entry = iter.next();
        String datanodeUuid = entry.getKey();
        LightWeightHashSet<Block> blocks = entry.getValue();
        // Sort blocks by timestamp in descending order.
        List<ExcessBlockInfo> sortedBlocks = blocks.stream()
            .filter(block -> block instanceof ExcessBlockInfo)
            .map(block -> (ExcessBlockInfo) block)
            .sorted(Comparator.comparingLong(ExcessBlockInfo::getTimeStamp))
            .collect(Collectors.toList());

        for (ExcessBlockInfo excessBlockInfo : sortedBlocks) {
          if (processed >= excessRedundancyTimeoutCheckLimit) {
            break;
          }

          processed++;
          // If the datanode doesn't have any excess block that has exceeded the timeout,
          // can exit this loop.
          if (now <= excessBlockInfo.getTimeStamp() + excessRedundancyTimeout) {
            break;
          }

          BlockInfo blockInfo = excessBlockInfo.getBlockInfo();
          BlockInfo bi = blocksMap.getStoredBlock(blockInfo);
          if (bi == null || bi.isDeleted()) {
            continue;
          }

          Iterator<DatanodeStorageInfo> iterator = blockInfo.getStorageInfos();
          while (iterator.hasNext()) {
            DatanodeStorageInfo datanodeStorageInfo = iterator.next();
            DatanodeDescriptor datanodeDescriptor = datanodeStorageInfo.getDatanodeDescriptor();
            if (datanodeDescriptor.getDatanodeUuid().equals(datanodeUuid) &&
                datanodeStorageInfo.getState().equals(State.NORMAL)) {
              final Block b = getBlockOnStorage(blockInfo, datanodeStorageInfo);
              if (!containsInvalidateBlock(datanodeDescriptor, b)) {
                addToInvalidates(b, datanodeDescriptor);
                LOG.debug("Excess block timeout ({}, {}) is added to invalidated.",
                    b, datanodeDescriptor);
              }
              excessBlockInfo.setTimeStamp();
              break;
            }
          }
        }
      }
    } finally {
      namesystem.writeUnlock("processTimedOutExcessBlocks");
      LOG.info("processTimedOutExcessBlocks {} msecs.", (Time.monotonicNow() - now));
    }
  }
@Test(timeout = 360000)
  public void testProcessTimedOutExcessBlocks() throws IOException,
      InterruptedException, TimeoutException {
    Configuration config = new HdfsConfiguration();
    // Bump up replication interval.
    config.setInt(DFSConfigKeys.DFS_NAMENODE_REDUNDANCY_INTERVAL_SECONDS_KEY, 10000);
    // Set the excess redundancy block timeout.
    long timeOut = 60L;
    config.setLong(DFSConfigKeys.DFS_NAMENODE_EXCESS_REDUNDANCY_TIMEOUT_SEC_KEY, timeOut);

    DataNodeFaultInjector oldInjector = DataNodeFaultInjector.get();

    final Semaphore semaphore = new Semaphore(0);
    try (MiniDFSCluster cluster = new MiniDFSCluster.Builder(config).numDataNodes(3).build()) {
      DistributedFileSystem fs = cluster.getFileSystem();
      BlockManager blockManager = cluster.getNameNode().getNamesystem().getBlockManager();
      cluster.waitActive();

      final DataNodeFaultInjector injector = new DataNodeFaultInjector() {
        @Override
        public void delayDeleteReplica() {
          // Lets wait for the remove replica process.
          try {
            semaphore.acquire(1);
          } catch (InterruptedException e) {
            // ignore.
          }
        }
      };
      DataNodeFaultInjector.set(injector);

      // Create file.
      Path path = new Path("/testfile");
      DFSTestUtil.createFile(fs, path, 1024, (short) 3, 0);
      DFSTestUtil.waitReplication(fs, path, (short) 3);
      LocatedBlock lb = DFSTestUtil.getAllBlocks(fs, path).get(0);
      ExtendedBlock extendedBlock = lb.getBlock();
      DatanodeInfo[] loc = lb.getLocations();
      assertEquals(3, loc.length);

      // Set replication as 2, to choose excess.
      fs.setReplication(path, (short) 2);

      // Check excessRedundancyMap and invalidateBlocks size as 1.
      assertEquals(1, blockManager.getExcessBlocksCount());
      assertEquals(1, blockManager.getPendingDeletionBlocksCount());
      DataNode excessDn = Arrays.stream(loc).
          filter(datanodeInfo -> blockManager.getExcessSize4Testing(
              datanodeInfo.getDatanodeUuid()) > 0)
          .map(datanodeInfo -> cluster.getDataNode(datanodeInfo.getIpcPort()))
          .findFirst()
          .orElse(null);

      // Schedule blocks for deletion at excessDn.
      assertEquals(1, blockManager.computeInvalidateWork(1));
      // Check excessRedundancyMap size as 1.
      assertEquals(1, blockManager.getExcessBlocksCount());
      // Check invalidateBlocks size as 0.
      assertEquals(0, blockManager.getPendingDeletionBlocksCount());
      assertNotNull(excessDn);

      // NameNode will ask datanode to delete replicas in heartbeat response.
      cluster.triggerHeartbeats();

      // Wait for the datanode to process any block deletions
      // that have already been asynchronously queued.
      DataNode finalExcessDn = excessDn;
      GenericTestUtils.waitFor(
          () -> cluster.getFsDatasetTestUtils(finalExcessDn).getPendingAsyncDeletions() == 1,
          100, 1000);

      // Restart the datanode.
      int ipcPort = excessDn.getDatanodeId().getIpcPort();
      MiniDFSCluster.DataNodeProperties dataNodeProperties = cluster.stopDataNode(
          excessDn.getDatanodeId().getXferAddr());
      assertTrue(cluster.restartDataNode(dataNodeProperties, true));
      semaphore.release(1);
      cluster.waitActive();

      // Check replica is exists in excessDn.
      excessDn = cluster.getDataNode(ipcPort);
      assertNotNull(cluster.getFsDatasetTestUtils(excessDn).fetchReplica(extendedBlock));
      assertEquals(0, cluster.getFsDatasetTestUtils(excessDn).getPendingAsyncDeletions());

      // Verify excess redundancy blocks have not timed out.
      blockManager.processTimedOutExcessBlocks();
      assertEquals(0, blockManager.getPendingDeletionBlocksCount());

      // Verify excess redundancy block time out.
      Thread.sleep(timeOut * 1000);
      blockManager.processTimedOutExcessBlocks();

      // Check excessRedundancyMap and invalidateBlocks size as 1.
      assertEquals(1, blockManager.getExcessSize4Testing(excessDn.getDatanodeUuid()));
      assertEquals(1, blockManager.getExcessBlocksCount());
      assertEquals(1, blockManager.getPendingDeletionBlocksCount());

      // Schedule blocks for deletion.
      assertEquals(1, blockManager.computeInvalidateWork(1));

      cluster.triggerHeartbeats();

      // Make it resume the removeReplicaFromMem method.
      semaphore.release(1);

      // Wait for the datanode in the cluster to process any block
      // deletions that have already been asynchronously queued
      cluster.waitForDNDeletions();

      // Trigger immediate deletion report.
      cluster.triggerDeletionReports();

      // The replica num should be 2.
      assertEquals(2, DFSTestUtil.getAllBlocks(fs, path).get(0).getLocations().length);
      assertEquals(0, blockManager.getExcessBlocksCount());
    } finally {
      DataNodeFaultInjector.set(oldInjector);
    }
  }