@Test
public void testDynamicDistCp() throws Exception {

  try {
    clearState();
    Path sourcePath = new Path(SOURCE_PATH)
            .makeQualified(cluster.getFileSystem());
    List<Path> sources = new ArrayList<Path>();
    sources.add(sourcePath);
    Path targetPath = new Path(TARGET_PATH)
            .makeQualified(cluster.getFileSystem());
    DistCpOptions options = new DistCpOptions(sources, targetPath);
    options.setCopyStrategy("dynamic");
    options.setOutPutDirectory(counterOutputPath);
    options.setAtomicCommit(true);
    options.setAtomicWorkPath(new Path("/work"));
    options.setBlocking(false);
    Job job = new DistCp(configuration, options).execute();
    Path workDir = CopyOutputFormat.getWorkingDirectory(job);
    Path finalDir = CopyOutputFormat.getCommitDirectory(job);
    while (!job.isComplete()) {
      if (cluster.getFileSystem().exists(workDir)) {
        break;
      }
    }
    job.waitForCompletion(true);
    Assert.assertFalse(cluster.getFileSystem().exists(workDir));
    Assert.assertTrue(cluster.getFileSystem().exists(finalDir));
    verifyResults();
  }
  catch (Exception e) {
    LOG.error("Exception encountered", e);
    Assert.fail("Unexpected exception: " + e.getMessage());
  }
}