@Override
  public void setOffset(ReplicationQueueId queueId, String walGroup, ReplicationGroupOffset offset,
    Map<String, Long> lastSeqIds) throws ReplicationException {
    Put put = new Put(Bytes.toBytes(queueId.toString())).addColumn(QUEUE_FAMILY,
      Bytes.toBytes(walGroup), Bytes.toBytes(offset.toString()));
    AsyncTable<?> asyncTable = conn.toAsyncConnection().getTable(tableName);
    try {
      if (lastSeqIds.isEmpty()) {
        FutureUtils.get(asyncTable.put(put));
      } else {
        for (;;) {
          MultiRowMutationProtos.MutateRowsRequest.Builder builder =
            MultiRowMutationProtos.MutateRowsRequest.newBuilder();
          addLastSeqIdsPut(builder, queueId.getPeerId(), lastSeqIds, asyncTable);
          if (builder.getMutationRequestCount() > 0) {
            // use MultiRowMutationService to atomically update offset and last sequence ids
            MultiRowMutationProtos.MutateRowsRequest request =
              builder.addMutationRequest(ProtobufUtil.toMutation(MutationType.PUT, put)).build();
            MultiRowMutationProtos.MutateRowsResponse responose =
              FutureUtils.get(asyncTable.<MultiRowMutationProtos.MultiRowMutationService.Interface,
                MultiRowMutationProtos.MutateRowsResponse> coprocessorService(
                  MultiRowMutationProtos.MultiRowMutationService::newStub,
                  (stub, controller, done) -> stub.mutateRows(controller, request, done),
                  put.getRow()));
            if (responose.getProcessed()) {
              break;
            }
          } else {
            // we do not need to update last seq id, fallback to single put
            FutureUtils.get(asyncTable.put(put));
            break;
          }
        }
      }
    } catch (IOException e) {
      throw new ReplicationException("failed to setOffset, queueId=" + queueId + ", walGroup="
        + walGroup + ", offset=" + offset + ", lastSeqIds=" + lastSeqIds, e);
    }
  }
HBase table based replication queue storage.
hbase-replication/src/main/java/org/apache/hadoop/hbase/replication/TableReplicationQueueStorage.java