@Override
  public void copy(@Nonnull FileCopyInfo fileCopyInfo) throws Exception {
    logger.info("File Copy handler is running for partition: {} on Mount Path: {} hydrating from DataNode: {}",
        fileCopyInfo.getSourceReplicaId().getPartitionId().getId(),
        fileCopyInfo.getSourceReplicaId().getMountPath(), fileCopyInfo.getTargetReplicaId().getDataNodeId());
    Objects.requireNonNull(fileCopyInfo, "fileCopyReplicaInfo param cannot be null");
    validateIfStoreFileCopyHandlerIsRunning();

    final PartitionFileStore fileStore = storeManager.getFileStore(fileCopyInfo.getSourceReplicaId().getPartitionId());
    final String partitionToMountFilePath = fileCopyInfo.getSourceReplicaId().getMountPath() + File.separator +
        fileCopyInfo.getSourceReplicaId().getPartitionId().getId();
    FileCopyGetMetaDataResponse metadataResponse = null;
    try {
      metadataResponse = getFileCopyGetMetaDataResponse(fileCopyInfo);
      String snapshotId = metadataResponse.getSnapshotId();

      metadataResponse.getLogInfos().forEach(logInfo -> {
        // Process the respective files and copy it to the temporary path.
        final String partitionToMountTempFilePath = partitionToMountFilePath + File.separator + storeConfig.storeFileCopyTemporaryDirectoryName;
        logInfo.getIndexSegments().forEach(indexFile -> {
          processIndexFile(indexFile, partitionToMountTempFilePath, fileCopyInfo, snapshotId, fileStore);
        });
        // Process log segment
        long storeId = fileCopyInfo.getSourceReplicaId().getPartitionId().getId();
        // allocate the file
        FileInfo logFileInfo = new StoreFileInfo(logInfo.getLogSegment().getFileName() + "_log",
            logInfo.getLogSegment().getFileSize());
        String filePath = partitionToMountTempFilePath + File.separator + logFileInfo.getFileName();
        try {
          fileStore.allocateFile(filePath, Long.toString(storeId));
        } catch (IOException e) {
          logMessageAndThrow("ProcessLogSegment", "Failed Disk Space Allocation", e,
              FileCopyHandlerException.FileCopyHandlerErrorCode.FileCopyHandlerFailedDiskSpaceAllocation);
        }
        processLogSegment(logInfo, partitionToMountTempFilePath, fileCopyInfo, snapshotId, fileStore);

        // Move all files to actual path.
        try {
          fileStore.moveAllRegularFiles(partitionToMountTempFilePath, partitionToMountFilePath);
        } catch (IOException e) {
          logMessageAndThrow("MoveFilesOperation", "Error moving files", e,
              FileCopyHandlerException.FileCopyHandlerErrorCode.FileCopyHandlerWriteToDiskError);
        }
      });
    } finally {
      if (metadataResponse != null) {
        metadataResponse.release();
      }
    }
  }
Copy the file from the source replica to the destination replica.
@param fileCopyInfo the replica info of type {@link FileCopyInfo}
@throws IOException
ambry-file-transfer/src/main/java/com/github/ambry/filetransfer/handler/StoreFileCopyHandler.java