@Override
  public Map<DataNodeId, List<RemoteReplicaInfo>> selectReplicas(Map<DataNodeId, List<RemoteReplicaInfo>> replicas) {
    HashMap<Long, ArrayList<RemoteReplicaInfo>> partitions = new HashMap<>();
    // Group replicas by partition
    replicas.values().forEach(rlist -> rlist.forEach(replica -> partitions.computeIfAbsent(
        replica.getReplicaId().getPartitionId().getId(), k -> new ArrayList<>()).add(replica)));
    // Pick one replica per partition for this iteration
    Map<DataNodeId, List<RemoteReplicaInfo>> nodes = new HashMap<>();
    partitions.values().forEach(rlist -> {
      rlist.sort(comparator);
      RemoteReplicaInfo replica;
      switch (vcrNodeConfig.replicaSelectionPolicy) {
        case FIXED:
          replica = rlist.get(0);
          logger.trace("FIXED replicaSelectionPolicy picked {} for partition-{}",
              replica.getReplicaId().getDataNodeId().getHostname(), replica.getReplicaId().getPartitionId().getId());
          break;
        case ROUND_ROBIN:
        default:
          replica = rlist.get(numReplIter % rlist.size());
          logger.trace("{} replicaSelectionPolicy picked {} for partition-{}",
              vcrNodeConfig.DEFAULT_REPLICA_SELECTION_POLICY, replica.getReplicaId().getDataNodeId().getHostname(),
              replica.getReplicaId().getPartitionId().getId());
      }
      if (replica.getToken() == null) {
        setReplicaToken(replica);
      }
      // Group by data node
      nodes.computeIfAbsent(replica.getReplicaId().getDataNodeId(), k -> new ArrayList<>()).add(replica);
    });
    numReplIter = (numReplIter % 100) + 1; // Prevent integer overflow
    return nodes;
  }
Selects replicas R1, R2, and R3 of a partition P in distinct iterations of the replication loop.
As the loop is continuous, each replica gets its turn.
Consequently, in the first iteration, R1 is processed, followed by R2, and then R3.
This approach ensures that missing blob B is uploaded from R1 to Azure during the first iteration.
Subsequent iterations involving R2 and R3 skip the fetch and upload step for blob B, as it is already present.

There sure is a better algorithm to do this but the size of the input is enough for this crude code.
@param replicas A map of replicas {host -> {replicas}}
ambry-cloud/src/main/java/com/github/ambry/cloud/VcrReplicaThread.java