@Override
  public void scheduleFileCopy() throws InterruptedException {
    logger.info("Starting File Copy Scheduler");
    while(isRunning){

      Thread.sleep(fileCopyBasedReplicationConfig.fileCopySchedulerWaitTimeSecs*1000);

      List<ReplicaId> replicasToDropForHydration = findStarvedReplicas();
      if(!replicasToDropForHydration.isEmpty()){
        logger.info("Found Replicas To Drop From Hydration: " + replicasToDropForHydration.stream()
            .map(replicaId -> replicaId.getPartitionId().toPathString()).collect(Collectors.toList()));
      } else{
        logger.info("No Replicas To Drop From Hydration In Current Cycle");
      }

      for(ReplicaId replica: replicasToDropForHydration){
        try {
          fileCopyBasedReplicationThreadPoolManager.stopAndRemoveReplicaFromThreadPool(replica);
        } catch (InterruptedException e) {
          //TODO: Send Alert On Failure
          logger.error("Error Stopping Replica: " + replica.getPartitionId().toPathString());
          logger.error("[Error]: ", e);
        }
        replicaToStartTimeMap.remove(replica);
        replicaToStatusListenerMap.get(replica).onFileCopyFailure(new Exception("Replica Timed Out"));
        replicaToStatusListenerMap.remove(replica);
        inFlightReplicas.remove(replica);
        // TODO: The Replica should be sent to the end of the queue inside prioritization Manager
        //  instead of removing it to give other replicas a chance.
        prioritizationManager.removeReplica(replica.getDiskId(), replica);
      }

      List<DiskId> disksToHydrate = fileCopyBasedReplicationThreadPoolManager.getDiskIdsToHydrate();
      for(DiskId diskId: disksToHydrate){
        List<ReplicaId> replicaIds = getNextReplicaToHydrate(diskId, fileCopyBasedReplicationConfig.fileCopyParallelPartitionHydrationCountPerDisk);

        if(!replicaIds.isEmpty()){
          logger.info("Starting Hydration For Disk: {}", diskId);
          for(ReplicaId replicaId: replicaIds) {
            if (inFlightReplicas.contains(replicaId)) {
              continue;
            }
            FileCopyStatusListener fileCopyStatusListener = new FileCopyStatusListenerImpl(replicaSyncUpManager, replicaId);
            FileCopyHandler fileCopyHandler = fileCopyHandlerFactory.getFileCopyHandler();
            try{
              /**
               * Use FileCopyTemporaryDirectoryName to create a temporary directory for file copy.
               * This will be used to write the files which are not yet written and can be cleaned
               * up without
               */
              createTemporaryDirectoryForFileCopyIfAbsent(replicaId, storeConfig);
            } catch (IOException e){
              logger.error("Error Creating Temporary Directory For Replica: " + replicaId.getPartitionId().toPathString());
              fileCopyStatusListener.onFileCopyFailure(e);
              continue;
            }

            fileCopyBasedReplicationThreadPoolManager.submitReplicaForHydration(replicaId,
                fileCopyStatusListener, fileCopyHandler);

            replicaToStatusListenerMap.put(replicaId, fileCopyStatusListener);
            inFlightReplicas.add(replicaId);
            replicaToStartTimeMap.put(replicaId, System.currentTimeMillis()/1000);
          }
        } else{
          logger.info("No Replicas To Hydrate For Disk: " + diskId);
        }
      }
    }

    logger.error("FileCopyBasedReplicationSchedulerImpl Stopped");
  }
Find the replicas that are starved for hydration. Return a list of Replicas
which are stuck in hydration for more than the configured values.
@return the list of {@link ReplicaId} that are starved for hydration.
ambry-file-transfer/src/main/java/com/github/ambry/filetransfer/FileCopyBasedReplicationSchedulerImpl.java