private long getStartOffset(String walGroupId) {
    ReplicationGroupOffset startOffset = startOffsets.get(walGroupId);
    if (startOffset == null || startOffset == ReplicationGroupOffset.BEGIN) {
      return 0L;
    }
    // this method will only be called when start new shipper, and we will only start new shipper
    // when there is a new queue, so here the queue for walGroupId will never be null.
    Path first = logQueue.getQueue(walGroupId).peek();
    if (!startOffset.getWal().equals(first.getName())) {
      return 0L;
    }
    // Usually, if we arrive here, the start offset should never be -1, as it means this file has
    // been fully replicated so we should have filtered it out in upper layer, usually in
    // ReplicationSourceManager. Add a warn message for safety, as usually replicate more data will
    // not cause big problems.
    if (startOffset.getOffset() < 0) {
      LOG.warn("Should have already replicated wal {}, return start offset as 0",
        startOffset.getWal());
      return 0L;
    } else {
      return startOffset.getOffset();
    }
  }
Instantiation method used by region servers
@param conf      configuration to use
@param fs        file system to use
@param manager   replication manager to ping to
@param server    the server for this region server
@param queueData the id and offsets of our replication queue
@param clusterId unique UUID for the cluster
@param metrics   metrics for replication source
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java