public void replicateContinuous() {
    long oneRoundStartTimeMs = time.milliseconds();
    terminateCurrentContinuousReplicationCycle = false;
    logger.trace("Thread name: {} Start RemoteReplicaGroup replication", threadName);

    exchangeMetadataResponsesInEachCycle = new HashMap<>();
    Set<Integer> groupsAddedToExchangeMetadatResponse = new HashSet<>();
    boolean allReplicasCaughtEarly = false;

    try {
      storeKeyConverter.dropCache();

      Map<Integer, RemoteReplicaGroup> correlationIdToReplicaGroup = new HashMap<>();
      Map<Integer, RequestInfo> correlationIdToRequestInfo = new LinkedHashMap<>();

      RemoteReplicaGroupPoller poller = new RemoteReplicaGroupPoller();
      // we keep polling groups until we can, i.e. there are no more inflight groups

      do {
        if (!running) {
          break;
        }
        List<RemoteReplicaGroup> inflightRemoteReplicaGroups = poller.enqueue();

        pollRequestsAndSubmitResponses(inflightRemoteReplicaGroups, correlationIdToReplicaGroup,
            correlationIdToRequestInfo);

        processFinishedGroups(inflightRemoteReplicaGroups, groupsAddedToExchangeMetadatResponse);
        logger.trace("Thread name: {} Finish one iteration for RemoteReplicaGroup replication", threadName);
      } while (poller.shouldContinue());
      logger.trace("Thread name: {} Finish all RemoteReplicaGroup replication", threadName);
      allReplicasCaughtEarly = poller.allReplicasCaughtUpEarly();
    } catch (Exception e) {
      logger.error("Thread name: {} found some error while replicating from remote hosts", threadName, e);
    } finally {
      long cycleEndTime = time.milliseconds();
      replicationMetrics.updateOneCycleReplicationTime(cycleEndTime - oneRoundStartTimeMs,
          replicatingFromRemoteColo, datacenterName);
      updatePerReplicationCycleTimeMetric(cycleEndTime - oneRoundStartTimeMs);
      updateReplicationCycleIterationsMetric();
    }

    // check and make thread sleep and publish metrics for throttling
    maybeSleepAfterReplication(allReplicasCaughtEarly);
    logger.trace("Thread name: {} Exiting replication, all iterations Done!", threadName);
  }
Do replication for replicas grouped by {@link DataNodeId}
We will keep trying to generate groups until we do not get any groups by calling generateRemoteReplicaGroups.
We will generate requests and then process responses when received for these groups.

A replication cycle between two replicas involves the following steps:
   1. Exchange metadata : fetch the metadata of blobs added to remote replica since the last synchronization point
   and filter the ones missing in local store.
   2. Fetch missing blobs: fetch the missing blobs by issuing GET request to remote replica and write them to
      the local store

 During cross-colo replication, depending on the {@link ReplicationModelType}, the missing blobs are either fetched
 from all remote replicas (if modelType == ALL_TO_ALL) or only fetched for local leader replicas from their remote
 leader replicas (if modelType == LEADER_BASED). In the latter case, non-leader replica pairs (leader <-> standby,
 standby <-> leader, standby <-> standby) will get their missing blobs from their corresponding leader<->leader
 exchanges and intra-dc replication.

 Here is a table listing on what is exchanged between local and remote replicas based on their roles
 (leader/standby) when {@link ReplicationModelType is LEADER_BASED}.

             |   Local Leader    |     Local Standby   |   Remote Leader   |  Remote Standby
           -------------------------------------------------------------------------------------
    Leader:  |        ---        |  metadata and data  | metadata and data |   metadata only
    Standby: | metadata and data |  metadata and data  | metadata only     |   metadata only
ambry-replication/src/main/java/com/github/ambry/replication/ReplicaThread.java