void claimQueue(ReplicationQueueId queueId, boolean syncUp) {
    // Wait a bit before transferring the queues, we may be shutting down.
    // This sleep may not be enough in some cases.
    try {
      Thread.sleep(sleepBeforeFailover
        + (long) (ThreadLocalRandom.current().nextFloat() * sleepBeforeFailover));
    } catch (InterruptedException e) {
      LOG.warn("Interrupted while waiting before transferring a queue.");
      Thread.currentThread().interrupt();
    }
    // We try to lock that rs' queue directory
    if (server.isStopped()) {
      LOG.info("Not transferring queue since we are shutting down");
      return;
    }
    // After claim the queues from dead region server, we will skip to start the
    // RecoveredReplicationSource if the peer has been removed. but there's possible that remove a
    // peer with peerId = 2 and add a peer with peerId = 2 again during failover. So we need to get
    // a copy of the replication peer first to decide whether we should start the
    // RecoveredReplicationSource. If the latest peer is not the old peer, we should also skip to
    // start the RecoveredReplicationSource, Otherwise the rs will abort (See HBASE-20475).
    String peerId = queueId.getPeerId();
    ReplicationPeerImpl oldPeer = replicationPeers.getPeer(peerId);
    if (oldPeer == null) {
      LOG.info("Not transferring queue since the replication peer {} for queue {} does not exist",
        peerId, queueId);
      return;
    }
    Map<String, ReplicationGroupOffset> offsets;
    try {
      offsets = queueStorage.claimQueue(queueId, server.getServerName());
    } catch (ReplicationException e) {
      LOG.error("ReplicationException: cannot claim dead region ({})'s replication queue",
        queueId.getServerName(), e);
      server.abort("Failed to claim queue from dead regionserver.", e);
      return;
    }
    if (offsets.isEmpty()) {
      // someone else claimed the queue
      return;
    }
    ServerName sourceRS = queueId.getServerWALsBelongTo();
    ReplicationQueueId claimedQueueId = queueId.claim(server.getServerName());
    ReplicationPeerImpl peer = replicationPeers.getPeer(peerId);
    if (peer == null || peer != oldPeer) {
      LOG.warn("Skipping failover for peer {} of node {}, peer is null", peerId, sourceRS);
      deleteQueue(claimedQueueId);
      return;
    }
    ReplicationSourceInterface src;
    try {
      src =
        createSource(new ReplicationQueueData(claimedQueueId, ImmutableMap.copyOf(offsets)), peer);
    } catch (IOException e) {
      LOG.error("Can not create replication source for peer {} and queue {}", peerId,
        claimedQueueId, e);
      server.abort("Failed to create replication source after claiming queue.", e);
      return;
    }
    PriorityQueue<Path> walFiles;
    try {
      walFiles = getWALFilesToReplicate(sourceRS, syncUp, offsets);
    } catch (IOException e) {
      LOG.error("Can not list wal files for peer {} and queue {}", peerId, queueId, e);
      server.abort("Can not list wal files after claiming queue.", e);
      return;
    }
    // synchronized on oldsources to avoid adding recovered source for the to-be-removed peer
    synchronized (oldsources) {
      addRecoveredSource(src, oldPeer, claimedQueueId, walFiles);
    }
  }
Claim a replication queue.
<p/>
We add a flag to indicate whether we are called by ReplicationSyncUp. For normal claiming queue
operation, we are the last step of a SCP, so we can assume that all the WAL files are under
oldWALs directory. But for ReplicationSyncUp, we may want to claim the replication queue for a
region server which has not been processed by SCP yet, so we still need to look at its WALs
directory.
@param queueId the replication queue id we want to claim
@param syncUp  whether we are called by ReplicationSyncUp
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java