### Question:
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); } } 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); @Override Metrics copy();  }### Answer:
@Test public void typical() throws Exception { copier = new DistCpCopier(conf, sourceDataBaseLocation, sourceDataLocations, replicaDataLocation, null, registry); Metrics metrics = copier.copy(); assertThat(metrics, not(nullValue())); String outputPath = replicaDataLocation.toUri().getPath(); File outputSub2Data = new File(outputPath, "sub1/sub2/data"); assertTrue(outputSub2Data.exists()); assertThat(Files.asCharSource(outputSub2Data, UTF_8).read(), is("test1")); File outputSub4Data = new File(outputPath, "sub3/sub4/data"); assertTrue(outputSub4Data.exists()); assertThat(Files.asCharSource(outputSub4Data, UTF_8).read(), is("test2")); assertThat(registry.getGauges().containsKey(RunningMetrics.DIST_CP_BYTES_REPLICATED.name()), is(true)); }
@Test public void typicalOneFile() throws Exception { Path inputFile = new Path(sourceDataBaseLocation, "sub1/sub2/data"); Path targetFile = new Path(replicaDataLocation, "output.txt"); copier = new DistCpCopier(conf, inputFile, Collections.<Path>emptyList(), targetFile, null, registry); Metrics metrics = copier.copy(); assertThat(metrics, not(nullValue())); String outputPath = targetFile.toUri().getPath(); Path parent = targetFile.getParent(); FileSystem fs = parent.getFileSystem(conf); int fileCopyCount = fs.listStatus(parent).length; assertThat(fileCopyCount, is(1)); File outputSub2Data = new File(outputPath); assertTrue(outputSub2Data.exists()); assertThat(Files.asCharSource(outputSub2Data, UTF_8).read(), is("test1")); }
@Test public void cleanUpOnFailure() throws Exception { Map<String, Object> copierOptions = new HashMap<>(); copierOptions.put("file-attribute", "xattr"); copier = new DistCpCopier(conf, sourceDataBaseLocation, sourceDataLocations, replicaDataLocation, copierOptions, registry); try { copier.copy(); fail("Expecting copier failure"); } catch (CircusTrainException e) { String outputPath = replicaDataLocation.toUri().toString(); assertFalse(new File(outputPath).exists()); } }