CopyCommitter extends FileOutputCommitter { 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(); 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)); 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)); 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; 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)) { 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; if (tracker.shouldDelete(trgtFileStatus)) { showProgress = true; try { if (targetFS.delete(targetEntry, true)) { LOG.info("Deleted " + targetEntry + " - missing at source"); deletedEntries++; if (trgtFileStatus.isDirectory()) { deletedDirectories++; } else { filesDeleted++; } } else { LOG.info("delete({}) returned false ({})", targetEntry, trgtFileStatus); missingDeletes++; } } catch (IOException e) { if (!ignoreFailures) { throw e; } else { LOG.info("Failed to delete {}, ignoring exception {}", targetEntry, e.toString()); LOG.debug("Failed to delete {}", targetEntry, e); failedDeletes++; } } } else { LOG.debug("Skipping deletion of {}", targetEntry); skippedDeletes++; showProgress = false; } if (showProgress) { taskAttemptContext.progress(); taskAttemptContext.setStatus("Deleting removed files from target. [" + targetReader.getPosition() * 100 / totalLen + "%]"); } } 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)); } CopyCommitter(Path outputPath, TaskAttemptContext context); @Override void commitJob(JobContext jobContext); @Override void abortJob(JobContext jobContext,
                       JobStatus.State state);  }
@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(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); committer.commitJob(jobContext); verifyFoldersAreInSync(fs, targetBase, sourceBase); verifyFoldersAreInSync(fs, sourceBase, targetBase); } finally { TestDistCpUtils.delete(fs, "/tmp1"); conf.set(DistCpConstants.CONF_LABEL_DELETE_MISSING, "false"); } }