class DistCpCopier implements Copier {

  @Override
  public Metrics copy() throws CircusTrainException {
    LOG.info("Copying table data.");
    LOG.debug("Invoking DistCp: {} -> {}", sourceDataBaseLocation, replicaDataLocation);

    DistCpOptions distCpOptions = parseCopierOptions(copierOptions);
    LOG.debug("Invoking DistCp with options: {}", distCpOptions);

    CircusTrainCopyListing.setAsCopyListingClass(conf);
    CircusTrainCopyListing.setRootPath(conf, sourceDataBaseLocation);

    try {
      distCpOptions.setBlocking(false);
      Job job = executor.exec(conf, distCpOptions);
      String counter = String
          .format("%s_BYTES_WRITTEN", replicaDataLocation.toUri().getScheme().toUpperCase(Locale.ROOT));
      registerRunningJobMetrics(job, counter);
      if (!job.waitForCompletion(true)) {
        throw new IOException(
            "DistCp failure: Job " + job.getJobID() + " has failed: " + job.getStatus().getFailureInfo());
      }

      return new JobMetrics(job, FileSystemCounter.class.getName(), counter);
    } catch (Exception e) {
      cleanUpReplicaDataLocation();
      throw new CircusTrainException("Unable to copy file(s)", e);
    }
  }

  public  DistCpCopier(
      Configuration conf,
      Path sourceDataBaseLocation,
      List<Path> sourceDataLocations,
      Path replicaDataLocation,
      Map<String, Object> copierOptions,
      MetricRegistry registry);
    DistCpCopier(
      Configuration conf,
      Path sourceDataBaseLocation,
      List<Path> sourceDataLocations,
      Path replicaDataLocation,
      Map<String, Object> copierOptions,
      DistCpExecutor executor,
      MetricRegistry registry);

}