@Override
  public Map<String, ReplicationGroupOffset> claimQueue(ReplicationQueueId queueId,
    ServerName targetServerName) throws ReplicationException {
    ReplicationQueueId newQueueId = queueId.claim(targetServerName);
    byte[] coprocessorRow = ReplicationQueueId.getScanPrefix(queueId.getPeerId());
    AsyncTable<?> asyncTable = conn.toAsyncConnection().getTable(tableName);
    try (Table table = conn.getTable(tableName)) {
      for (;;) {
        Map<String, ReplicationGroupOffset> offsets = getOffsets0(table, queueId);
        if (offsets.isEmpty()) {
          return Collections.emptyMap();
        }
        Map.Entry<String, ReplicationGroupOffset> entry = offsets.entrySet().iterator().next();
        ClientProtos.Condition condition = ProtobufUtil.toCondition(
          Bytes.toBytes(queueId.toString()), QUEUE_FAMILY, Bytes.toBytes(entry.getKey()),
          CompareOperator.EQUAL, Bytes.toBytes(entry.getValue().toString()), null);
        Delete delete = new Delete(Bytes.toBytes(queueId.toString())).addFamily(QUEUE_FAMILY);
        Put put = new Put(Bytes.toBytes(newQueueId.toString()));
        offsets.forEach((walGroup, offset) -> put.addColumn(QUEUE_FAMILY, Bytes.toBytes(walGroup),
          Bytes.toBytes(offset.toString())));
        MultiRowMutationProtos.MutateRowsRequest request =
          MultiRowMutationProtos.MutateRowsRequest.newBuilder().addCondition(condition)
            .addMutationRequest(ProtobufUtil.toMutation(MutationType.DELETE, delete))
            .addMutationRequest(ProtobufUtil.toMutation(MutationType.PUT, put)).build();
        MultiRowMutationProtos.MutateRowsResponse resp =
          FutureUtils.get(asyncTable.<MultiRowMutationProtos.MultiRowMutationService.Interface,
            MultiRowMutationProtos.MutateRowsResponse> coprocessorService(
              MultiRowMutationProtos.MultiRowMutationService::newStub,
              (stub, controller, done) -> stub.mutateRows(controller, request, done),
              coprocessorRow));
        if (resp.getProcessed()) {
          return offsets;
        }
        // if the multi is not processed, which usually the queue has already been claimed by
        // others, for safety, let's try claiming again, usually the next get operation above will
        // return an empty map and we will quit the loop.
      }
    } catch (IOException e) {
      throw new ReplicationException(
        "failed to claimQueue, queueId=" + queueId + ", targetServerName=" + targetServerName, e);
    }
  }
HBase table based replication queue storage.
hbase-replication/src/main/java/org/apache/hadoop/hbase/replication/TableReplicationQueueStorage.java