void cleanOldLogs(String log, boolean inclusive, ReplicationSourceInterface source) {
    String logPrefix = AbstractFSWALProvider.getWALPrefixFromWALName(log);
    if (source.isRecovered()) {
      NavigableSet<Path> wals = walsByIdRecoveredQueues.get(source.getQueueId()).get(logPrefix);
      if (wals != null) {
        // here we just want to compare the timestamp, so it is OK to just create a fake WAL path
        NavigableSet<String> walsToRemove = wals.headSet(new Path(oldLogDir, log), inclusive)
          .stream().map(Path::getName).collect(Collectors.toCollection(TreeSet::new));
        if (walsToRemove.isEmpty()) {
          return;
        }
        cleanOldLogs(walsToRemove, source);
        walsToRemove.clear();
      }
    } else {
      NavigableSet<String> wals;
      NavigableSet<String> walsToRemove;
      // synchronized on walsById to avoid race with postLogRoll
      synchronized (this.walsById) {
        wals = walsById.get(source.getQueueId()).get(logPrefix);
        if (wals == null) {
          return;
        }
        walsToRemove = wals.headSet(log, inclusive);
        if (walsToRemove.isEmpty()) {
          return;
        }
        walsToRemove = new TreeSet<>(walsToRemove);
      }
      // cleanOldLogs may spend some time, especially for sync replication where we may want to
      // remove remote wals as the remote cluster may have already been down, so we do it outside
      // the lock to avoid block preLogRoll
      cleanOldLogs(walsToRemove, source);
      // now let's remove the files in the set
      synchronized (this.walsById) {
        wals.removeAll(walsToRemove);
      }
    }
  }
Cleans a log file and all older logs from replication queue. Called when we are sure that a log
file is closed and has no more entries.
@param log       Path to the log
@param inclusive whether we should also remove the given log file
@param source    the replication source
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java