ReplicationLogCleaner.java 文件源码

/**
 * Load all wals in all replication queues from ZK. This method guarantees to return a
 * snapshot which contains all WALs in the zookeeper at the start of this call even there
 * is concurrent queue failover. However, some newly created WALs during the call may
 * not be included.
 */
private Set<String> loadWALsFromQueues() throws KeeperException {
  for (int retry = 0; ; retry++) {
    int v0 = replicationQueues.getQueuesZNodeCversion();
    List<String> rss = replicationQueues.getListOfReplicators();
    if (rss == null) {
      LOG.debug("Didn't find any region server that replicates, won't prevent any deletions.");
      return ImmutableSet.of();
    }
    Set<String> wals = Sets.newHashSet();
    for (String rs : rss) {
      List<String> listOfPeers = replicationQueues.getAllQueues(rs);
      // if rs just died, this will be null
      if (listOfPeers == null) {
        continue;
      }
      for (String id : listOfPeers) {
        List<String> peersWals = replicationQueues.getLogsInQueue(rs, id);
        if (peersWals != null) {
          wals.addAll(peersWals);
        }
      }
    }
    int v1 = replicationQueues.getQueuesZNodeCversion();
    if (v0 == v1) {
      return wals;
    }
    LOG.info(String.format("Replication queue node cversion changed from %d to %d, retry = %d",
        v0, v1, retry));
  }
}