@Test
public void testCleanup() {

  try {
    clearState();
    Path sourcePath = new Path("noscheme:///file");
    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.setOutPutDirectory(counterOutputPath);
    Path stagingDir = null/* = JobSubmissionFiles.getStagingDir(
        new JobClient(new JobConf(configuration)), configuration)*/;
    boolean hadoop1 = false;
    try {
      LOG.info("Trying to get staging path using hadoop-2");
      Class clusterClass = DistCp.class.getClassLoader().loadClass(
          "org.apache.hadoop.mapreduce.Cluster");
      Method method = JobSubmissionFiles.class.getMethod("getStagingDir",
          clusterClass, Configuration.class);
      Constructor constructor = clusterClass.getConstructor(Configuration.class);
      stagingDir = (Path) method.invoke(null,
          constructor.newInstance(configuration), configuration);
    } catch (Exception ignored) {
      // fallback to hadoop-1 API
      hadoop1= true;
    }
    if (hadoop1) {
      try {
        LOG.info("Trying to get staging path using hadoop-1");
        Method method = JobSubmissionFiles.class.getMethod("getStagingDir",
            JobClient.class, Configuration.class);
        stagingDir = (Path) method.invoke(null,
            new JobClient(new JobConf(configuration)), configuration);
      } catch (Exception ignored) {
        // do nothing
      }
    }
    stagingDir.getFileSystem(configuration).mkdirs(stagingDir);
    try {
      new DistCp(configuration, options).execute();
    } catch (Throwable t) {
      Assert.assertEquals(stagingDir.getFileSystem(configuration).
          listStatus(stagingDir).length, 0);
    }
  } catch (Exception e) {
    LOG.error("Exception encountered ", e);
    Assert.fail("testCleanup failed " + e.getMessage());
  }
}