@Override
  public void commitJob(JobContext jobContext) throws IOException {
    Configuration conf = jobContext.getConfiguration();
    syncFolder = conf.getBoolean(DistCpConstants.CONF_LABEL_SYNC_FOLDERS, false);
    overwrite = conf.getBoolean(DistCpConstants.CONF_LABEL_OVERWRITE, false);
    updateRoot =
        conf.getBoolean(CONF_LABEL_UPDATE_ROOT, false);
    targetPathExists = conf.getBoolean(
        DistCpConstants.CONF_LABEL_TARGET_PATH_EXISTS, true);
    ignoreFailures = conf.getBoolean(
        DistCpOptionSwitch.IGNORE_FAILURES.getConfigLabel(), false);

    if (blocksPerChunk > 0) {
      concatFileChunks(conf);
    }

    super.commitJob(jobContext);

    cleanupTempFiles(jobContext);

    try {
      if (conf.getBoolean(DistCpConstants.CONF_LABEL_DELETE_MISSING, false)) {
        deleteMissing(conf);
      } else if (conf.getBoolean(DistCpConstants.CONF_LABEL_ATOMIC_COPY, false)) {
        commitData(conf);
      } else if (conf.get(CONF_LABEL_TRACK_MISSING) != null) {
        // save missing information to a directory
        trackMissing(conf);
      }
      // for HDFS-14621, should preserve status after -delete
      String attributes = conf.get(DistCpConstants.CONF_LABEL_PRESERVE_STATUS);
      final boolean preserveRawXattrs = conf.getBoolean(
              DistCpConstants.CONF_LABEL_PRESERVE_RAWXATTRS, false);
      if ((attributes != null && !attributes.isEmpty()) || preserveRawXattrs) {
        preserveFileAttributesForDirectories(conf);
      }
      taskAttemptContext.setStatus("Commit Successful");
    }
    finally {
      cleanup(conf);
    }
  }
@Test
  public void testAtomicCommitExistingFinal() throws IOException {
    TaskAttemptContext taskAttemptContext = getTaskAttemptContext(config);
    JobContext jobContext = new JobContextImpl(taskAttemptContext.getConfiguration(),
        taskAttemptContext.getTaskAttemptID().getJobID());
    Configuration conf = jobContext.getConfiguration();


    String workPath = "/tmp1/" + String.valueOf(rand.nextLong());
    String finalPath = "/tmp1/" + String.valueOf(rand.nextLong());
    FileSystem fs = null;
    try {
      OutputCommitter committer = new CopyCommitter(null, taskAttemptContext);
      fs = FileSystem.get(conf);
      fs.mkdirs(new Path(workPath));
      fs.mkdirs(new Path(finalPath));

      conf.set(CONF_LABEL_TARGET_WORK_PATH, workPath);
      conf.set(CONF_LABEL_TARGET_FINAL_PATH, finalPath);
      conf.setBoolean(DistCpConstants.CONF_LABEL_ATOMIC_COPY, true);

      assertPathExists(fs, "Work path", new Path(workPath));
      assertPathExists(fs, "Final path", new Path(finalPath));
      try {
        committer.commitJob(jobContext);
        Assert.fail("Should not be able to atomic-commit to pre-existing path.");
      } catch(Exception exception) {
        assertPathExists(fs, "Work path", new Path(workPath));
        assertPathExists(fs, "Final path", new Path(finalPath));
        LOG.info("Atomic-commit Test pass.");
      }

    } finally {
      TestDistCpUtils.delete(fs, workPath);
      TestDistCpUtils.delete(fs, finalPath);
      conf.setBoolean(DistCpConstants.CONF_LABEL_ATOMIC_COPY, false);
    }
  }