public RequestInfo poll() {
      Port port = remoteReplicaInfos.get(0).getPort();
      long timeout = replicationConfig.replicationRequestNetworkTimeoutMs;
      if (state == ReplicaGroupReplicationState.STARTED) {
        // When the state is STARTED, we will send the ReplicaMetadataRequest out.
        replicaMetadataRequestStartTimeMs = time.milliseconds();
        exchangeMetadataStartTimeInMs = replicaMetadataRequestStartTimeMs;
        ReplicaMetadataRequest request = createReplicaMetadataRequest(remoteReplicaInfos, remoteDataNode);
        RequestInfo requestInfo =
            new RequestInfo(remoteDataNode.getHostname(), port, request, remoteReplicaInfos.get(0).getReplicaId(), null,
                time.milliseconds(), timeout, timeout);
        logger.trace(
            "Remote node: {} Thread name: {} RemoteReplicaGroup {} Create ReplicaMetadataRequest, correlation id {}, state {}",
            remoteDataNode, threadName, id, requestInfo.getRequest().getCorrelationId(), state);
        state = ReplicaGroupReplicationState.REPLICA_METADATA_REQUEST_SENT;
        return requestInfo;
      } else if (state == ReplicaGroupReplicationState.REPLICA_METADATA_RESPONSE_HANDLED) {
        // When the state is REPLICA_METADATA_RESPONSE_HANDLED, we will send the GetRequest out if needed.
        remoteReplicaInfosToSend = remoteReplicaInfos;
        exchangeMetadataResponseListToProcess = exchangeMetadataResponseList;
        if (!isNonProgressStandbyReplicaGroup && replicatingFromRemoteColo && leaderBasedReplicationAdmin != null) {
          // If this is not standby replicas trying to catch up, and this is cross-colo replication, and we enable
          // leader based replication, then filter out remote replicas that is not leader to leader.
          List<RemoteReplicaInfo> leaderReplicaList = new ArrayList<>();
          List<ExchangeMetadataResponse> exchangeMetadataResponseListForLeaderReplicas = new ArrayList<>();
          getLeaderReplicaList(remoteReplicaInfos, exchangeMetadataResponseList, leaderReplicaList,
              exchangeMetadataResponseListForLeaderReplicas);
          remoteReplicaInfosToSend = leaderReplicaList;
          exchangeMetadataResponseListToProcess = exchangeMetadataResponseListForLeaderReplicas;
        }
        if (!remoteReplicaInfosToSend.isEmpty()) {
          if (exchangeMetadataResponseListToProcess.size() != remoteReplicaInfosToSend.size()) {
            throw new IllegalArgumentException("RemoteReplicaGroup " + id + " ExchangeMetadataResponseList size "
                + exchangeMetadataResponseListToProcess.size() + " and RemoteReplicaInfo size "
                + remoteReplicaInfosToSend.size() + " should be the same and greater than zero");
          }
          GetRequest request =
              createGetRequest(remoteReplicaInfosToSend, exchangeMetadataResponseListToProcess, remoteDataNode);
          if (request != null) {
            // Only when we still have RemoteReplicaInfos and when the GetRequest is not null
            // we send the GetRequest out
            getRequestStartTimeMs = time.milliseconds();
            fixMissingStoreKeysStartTimeInMs = getRequestStartTimeMs;
            RequestInfo requestInfo = new RequestInfo(remoteDataNode.getHostname(), port, request,
                remoteReplicaInfosToSend.get(0).getReplicaId(), null, time.milliseconds(), timeout, timeout);
            logger.trace(
                "Remote node: {} Thread name: {} RemoteReplicaGroup {} Create GetRequest, correlation id {}, state {}",
                remoteDataNode, threadName, id, requestInfo.getRequest().getCorrelationId(), state);
            state = ReplicaGroupReplicationState.GET_REQUEST_SENT;
            return requestInfo;
          }
        }
        logger.trace("Remote node: {} Thread name: {} RemoteReplicaGroup {} No GetRequest needed, set state to Done",
            remoteDataNode, threadName, id);
        state = ReplicaGroupReplicationState.DONE;
      }
      return null;
    }
Polling the {@link RequestInfo} from this group to send out. Each group has two requests to send out, one is
ReplicaMetadataRequest the other is GetRequest. Group only sends ReplicaMetadataRequest when it's in STARTED
state. And it only sends GetRequest when it's in REPLICA_METADATA_RESPONSE_HANDLED state. When group is not in
those states, this method would return null.

@return {@link RequestInfo} to send out. A null might be returned when there is nothing to send.
ambry-replication/src/main/java/com/github/ambry/replication/ReplicaThread.java