@Override
  public void preClean() {
    if (this.getConf() == null) {
      return;
    }
    try {
      if (!rpm.getQueueStorage().hasData()) {
        return;
      }
    } catch (ReplicationException e) {
      LOG.error("Error occurred while executing queueStorage.hasData()", e);
      return;
    }
    canFilter = barrier.start();
    if (canFilter) {
      notFullyDeadServers = getNotFullyDeadServers.get();
      peerIds = rpm.listPeers(null).stream().map(ReplicationPeerDescription::getPeerId)
        .collect(Collectors.toSet());
      // must get the not fully dead servers first and then get the replication queue data, in this
      // way we can make sure that, we should have added the missing replication queues for the dead
      // region servers recorded in the above set, otherwise the logic in the
      // filterForDeadRegionServer method may lead us delete wal still in use.
      List<ReplicationQueueData> allQueueData;
      try {
        allQueueData = rpm.getQueueStorage().listAllQueues();
      } catch (ReplicationException e) {
        LOG.error("Can not list all replication queues, give up cleaning", e);
        barrier.stop();
        canFilter = false;
        notFullyDeadServers = null;
        peerIds = null;
        return;
      }
      replicationOffsets = new HashMap<>();
      for (ReplicationQueueData queueData : allQueueData) {
        ReplicationQueueId queueId = queueData.getId();
        ServerName serverName = queueId.getServerWALsBelongTo();
        Map<String, Map<String, ReplicationGroupOffset>> peerId2Offsets =
          replicationOffsets.computeIfAbsent(serverName, k -> new HashMap<>());
        Map<String, ReplicationGroupOffset> offsets =
          peerId2Offsets.computeIfAbsent(queueId.getPeerId(), k -> new HashMap<>());
        offsets.putAll(queueData.getOffsets());
      }
    } else {
      LOG.info("Skip replication log cleaner because an AddPeerProcedure is running");
    }
  }
Implementation of a log cleaner that checks if a log is still scheduled for replication before
deleting it when its TTL is over.
<p/>
The logic is a bit complicated after we switch to use table based replication queue storage, see
the design doc in HBASE-27109 and the comments in HBASE-27214 for more details.
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/master/ReplicationLogCleaner.java