@Override
  protected Procedure<MasterProcedureEnv>[] execute(MasterProcedureEnv env)
    throws ProcedureYieldException, ProcedureSuspendedException, InterruptedException {
    ReplicationQueueStorage storage = env.getReplicationPeerManager().getQueueStorage();
    try {
      List<ReplicationQueueId> queues = storage.listAllQueueIds(crashedServer);
      Set<String> existQueuePeerIds = new HashSet<>();
      // this is for upgrading to the new region replication framework, where we will delete the
      // legacy region_replica_replication peer directly, without deleting the replication queues
      for (Iterator<ReplicationQueueId> iter = queues.iterator(); iter.hasNext();) {
        ReplicationQueueId queueId = iter.next();
        if (queueId.getPeerId().equals(ServerRegionReplicaUtil.REGION_REPLICA_REPLICATION_PEER)) {
          LOG.info("Found replication queue {} for legacy region replication peer, "
            + "skipping claiming and removing...", queueId);
          iter.remove();
          storage.removeQueue(queueId);
        } else if (!queueId.isRecovered()) {
          existQueuePeerIds.add(queueId.getPeerId());
        }
      }
      List<ReplicationPeerDescription> peers = env.getReplicationPeerManager().listPeers(null);
      // TODO: the implementation is not enough yet, if there are retries, we need to know whether
      // the replication queue for the given peer has been claimed or not, otherwise this logic will
      // introduce redundant replication queues for the same peer. Add this logic to make some UTs
      // pass first.
      for (ReplicationPeerDescription peer : peers) {
        if (!existQueuePeerIds.contains(peer.getPeerId())) {
          ReplicationQueueId queueId = new ReplicationQueueId(crashedServer, peer.getPeerId());
          env.getReplicationPeerManager().getQueueStorage().setOffset(queueId,
            crashedServer.toString(), ReplicationGroupOffset.BEGIN, Collections.emptyMap());
          queues.add(queueId);
        }
      }
      if (queues.isEmpty()) {
        LOG.debug("Finish claiming replication queues for {}", crashedServer);
        // we are done
        return null;
      }
      LOG.debug("There are {} replication queues need to be claimed for {}", queues.size(),
        crashedServer);
      List<ServerName> targetServers =
        env.getMasterServices().getServerManager().getOnlineServersList();
      if (targetServers.isEmpty()) {
        throw new ReplicationException("no region server available");
      }
      Collections.shuffle(targetServers);
      ClaimReplicationQueueRemoteProcedure[] procs =
        new ClaimReplicationQueueRemoteProcedure[Math.min(queues.size(), targetServers.size())];
      for (int i = 0; i < procs.length; i++) {
        procs[i] = new ClaimReplicationQueueRemoteProcedure(queues.get(i), targetServers.get(i));
      }
      return procs;
    } catch (ReplicationException e) {
      if (retryCounter == null) {
        retryCounter = ProcedureUtil.createRetryCounter(env.getMasterConfiguration());
      }
      long backoff = retryCounter.getBackoffTimeAndIncrementAttempts();
      LOG.warn("Failed to claim replication queues for {}, suspend {}secs {}; {};", crashedServer,
        backoff / 1000, e);
      setTimeout(Math.toIntExact(backoff));
      setState(ProcedureProtos.ProcedureState.WAITING_TIMEOUT);
      skipPersistence();
      throw new ProcedureSuspendedException();
    }
  }
Used to assign the replication queues of a dead server to other region servers.
@deprecated Use {@link AssignReplicationQueuesProcedure} instead, kept only for keeping
            compatibility.
hbase-server/src/main/java/org/apache/hadoop/hbase/master/replication/ClaimReplicationQueuesProcedure.java