private void deleteMissing(Configuration conf) throws IOException {
    LOG.info("-delete option is enabled. About to remove entries from " +
        "target that are missing in source");
    long listingStart = System.currentTimeMillis();

    // Sort the source-file listing alphabetically.
    Path sourceListing = new Path(conf.get(DistCpConstants.CONF_LABEL_LISTING_FILE_PATH));
    FileSystem clusterFS = sourceListing.getFileSystem(conf);
    Path sortedSourceListing = DistCpUtils.sortListing(conf, sourceListing);
    long sourceListingCompleted = System.currentTimeMillis();
    LOG.info("Source listing completed in {}",
        formatDuration(sourceListingCompleted - listingStart));

    // Similarly, create the listing of target-files. Sort alphabetically.
    Path targetListing = new Path(sourceListing.getParent(), "targetListing.seq");
    Path sortedTargetListing = new Path(targetListing.toString() + "_sorted");

    Path targetFinalPath = listTargetFiles(conf,
        targetListing, sortedTargetListing);
    long totalLen = clusterFS.getFileStatus(sortedTargetListing).getLen();

    SequenceFile.Reader sourceReader = new SequenceFile.Reader(conf,
                                 SequenceFile.Reader.file(sortedSourceListing));
    SequenceFile.Reader targetReader = new SequenceFile.Reader(conf,
                                 SequenceFile.Reader.file(sortedTargetListing));

    // Walk both source and target file listings.
    // Delete all from target that doesn't also exist on source.
    long deletionStart = System.currentTimeMillis();
    LOG.info("Destination listing completed in {}",
        formatDuration(deletionStart - sourceListingCompleted));

    long deletedEntries = 0;
    long filesDeleted = 0;
    long missingDeletes = 0;
    long failedDeletes = 0;
    long skippedDeletes = 0;
    long deletedDirectories = 0;
    // this is an arbitrary constant.
    final DeletedDirTracker tracker = new DeletedDirTracker(1000);
    try {
      CopyListingFileStatus srcFileStatus = new CopyListingFileStatus();
      Text srcRelPath = new Text();
      CopyListingFileStatus trgtFileStatus = new CopyListingFileStatus();
      Text trgtRelPath = new Text();

      final FileSystem targetFS = targetFinalPath.getFileSystem(conf);
      boolean showProgress;
      boolean srcAvailable = sourceReader.next(srcRelPath, srcFileStatus);
      while (targetReader.next(trgtRelPath, trgtFileStatus)) {
        // Skip sources that don't exist on target.
        while (srcAvailable && trgtRelPath.compareTo(srcRelPath) > 0) {
          srcAvailable = sourceReader.next(srcRelPath, srcFileStatus);
        }
        Path targetEntry = trgtFileStatus.getPath();
        LOG.debug("Comparing {} and {}",
            srcFileStatus.getPath(), targetEntry);

        if (srcAvailable && trgtRelPath.equals(srcRelPath)) continue;

        // Target doesn't exist at source. Try to delete it.
        if (tracker.shouldDelete(trgtFileStatus)) {
          showProgress = true;
          try {
            if (targetFS.delete(targetEntry, true)) {
              // the delete worked. Unless the file is actually missing, this is the
              LOG.info("Deleted " + targetEntry + " - missing at source");
              deletedEntries++;
              if (trgtFileStatus.isDirectory()) {
                deletedDirectories++;
              } else {
                filesDeleted++;
              }
            } else {
              // delete returned false.
              // For all the filestores which implement the FS spec properly,
              // this means "the file wasn't there".
              // so track but don't worry about it.
              LOG.info("delete({}) returned false ({})",
                  targetEntry, trgtFileStatus);
              missingDeletes++;
            }
          } catch (IOException e) {
            if (!ignoreFailures) {
              throw e;
            } else {
              // failed to delete, but ignoring errors. So continue
              LOG.info("Failed to delete {}, ignoring exception {}",
                  targetEntry, e.toString());
              LOG.debug("Failed to delete {}", targetEntry, e);
              // count and break out the loop
              failedDeletes++;
            }
          }
        } else {
          LOG.debug("Skipping deletion of {}", targetEntry);
          skippedDeletes++;
          showProgress = false;
        }
        if (showProgress) {
          // update progress if there's been any FS IO/files deleted.
          taskAttemptContext.progress();
          taskAttemptContext.setStatus("Deleting removed files from target. [" +
              targetReader.getPosition() * 100 / totalLen + "%]");
        }
      }
      // if the FS toString() call prints statistics, they get logged here
      LOG.info("Completed deletion of files from {}", targetFS);
    } finally {
      IOUtils.closeStream(sourceReader);
      IOUtils.closeStream(targetReader);
    }
    long deletionEnd = System.currentTimeMillis();
    long deletedFileCount = deletedEntries - deletedDirectories;
    LOG.info("Deleted from target: files: {} directories: {};"
            + " skipped deletions {}; deletions already missing {};"
            + " failed deletes {}",
        deletedFileCount, deletedDirectories, skippedDeletes,
        missingDeletes, failedDeletes);
    LOG.info("Number of tracked deleted directories {}", tracker.size());
    LOG.info("Duration of deletions: {}",
        formatDuration(deletionEnd - deletionStart));
    LOG.info("Total duration of deletion operation: {}",
        formatDuration(deletionEnd - listingStart));
  }
@Test
  public void testDeleteMissing() 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.createTestSetup(fs, FsPermission.getDefault());
      targetBase = TestDistCpUtils.createTestSetup(fs, FsPermission.getDefault());
      String targetBaseAdd = TestDistCpUtils.createTestSetup(fs, FsPermission.getDefault());
      fs.rename(new Path(targetBaseAdd), new Path(targetBase));

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

      CopyListing listing = new GlobbedCopyListing(conf, CREDENTIALS);
      Path listingFile = new Path("/tmp1/" + String.valueOf(rand.nextLong()));
      listing.buildListing(listingFile, context);

      conf.set(CONF_LABEL_TARGET_WORK_PATH, targetBase);
      conf.set(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");
    }
  }