@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 testDeleteMissingWithOnlyFile() throws IOException {
    TaskAttemptContext taskAttemptContext = getTaskAttemptContext(config);
    JobContext jobContext = new JobContextImpl(taskAttemptContext
        .getConfiguration(), taskAttemptContext.getTaskAttemptID().getJobID());
    Configuration conf = jobContext.getConfiguration();

    String sourceBase;
    String targetBase;
    FileSystem fs = null;
    try {
      OutputCommitter committer = new CopyCommitter(null, taskAttemptContext);
      fs = FileSystem.get(conf);
      sourceBase = TestDistCpUtils.createTestSetupWithOnlyFile(fs,
          FsPermission.getDefault());
      targetBase = TestDistCpUtils.createTestSetupWithOnlyFile(fs,
          FsPermission.getDefault());

      final DistCpOptions options = new DistCpOptions.Builder(
          Collections.singletonList(new Path(sourceBase)), new Path(targetBase))
          .withSyncFolder(true).withDeleteMissing(true).build();
      options.appendToConf(conf);
      final DistCpContext context = new DistCpContext(options);

      CopyListing listing = new GlobbedCopyListing(conf, CREDENTIALS);
      Path listingFile = new Path(sourceBase);
      listing.buildListing(listingFile, context);

      conf.set(DistCpConstants.CONF_LABEL_TARGET_WORK_PATH, targetBase);
      conf.set(DistCpConstants.CONF_LABEL_TARGET_FINAL_PATH, targetBase);

      committer.commitJob(jobContext);
      verifyFoldersAreInSync(fs, targetBase, sourceBase);
      verifyFoldersAreInSync(fs, sourceBase, targetBase);

      //Test for idempotent commit
      committer.commitJob(jobContext);
      verifyFoldersAreInSync(fs, targetBase, sourceBase);
      verifyFoldersAreInSync(fs, sourceBase, targetBase);
    } finally {
      TestDistCpUtils.delete(fs, "/tmp1");
      conf.set(DistCpConstants.CONF_LABEL_DELETE_MISSING, "false");
    }
  }