CopyCommitter extends FileOutputCommitter { @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); 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) { trackMissing(conf); } 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); } } CopyCommitter(Path outputPath, TaskAttemptContext context); @Override void commitJob(JobContext jobContext); @Override void abortJob(JobContext jobContext,
                       JobStatus.State state);  }
@Test public void testNoCommitAction() throws IOException { TaskAttemptContext taskAttemptContext = getTaskAttemptContext(config); JobContext jobContext = new JobContextImpl( taskAttemptContext.getConfiguration(), taskAttemptContext.getTaskAttemptID().getJobID()); OutputCommitter committer = new CopyCommitter(null, taskAttemptContext); committer.commitJob(jobContext); Assert.assertEquals("Commit Successful", taskAttemptContext.getStatus()); committer.commitJob(jobContext); Assert.assertEquals("Commit Successful", taskAttemptContext.getStatus()); }
@Test public void testPreserveStatus() 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); FsPermission sourcePerm = new FsPermission((short) 511); FsPermission initialPerm = new FsPermission((short) 448); sourceBase = TestDistCpUtils.createTestSetup(fs, sourcePerm); targetBase = TestDistCpUtils.createTestSetup(fs, initialPerm); final DistCpOptions options = new DistCpOptions.Builder( Collections.singletonList(new Path(sourceBase)), new Path("/out")) .preserve(FileAttribute.PERMISSION).build(); options.appendToConf(conf); final DistCpContext context = new DistCpContext(options); context.setTargetPathExists(false); CopyListing listing = new GlobbedCopyListing(conf, CREDENTIALS); Path listingFile = new Path("/tmp1/" + rand.nextLong()); listing.buildListing(listingFile, context); conf.set(CONF_LABEL_TARGET_FINAL_PATH, targetBase); committer.commitJob(jobContext); checkDirectoryPermissions(fs, targetBase, sourcePerm); committer.commitJob(jobContext); checkDirectoryPermissions(fs, targetBase, sourcePerm); } finally { TestDistCpUtils.delete(fs, "/tmp1"); conf.unset(DistCpConstants.CONF_LABEL_PRESERVE_STATUS); } }
@Test public void testPreserveStatusWithAtomicCommit() throws IOException { TaskAttemptContext taskAttemptContext = getTaskAttemptContext(config); JobContext jobContext = new JobContextImpl( taskAttemptContext.getConfiguration(), taskAttemptContext.getTaskAttemptID().getJobID()); Configuration conf = jobContext.getConfiguration(); String sourceBase; String workBase; String targetBase; FileSystem fs = null; try { OutputCommitter committer = new CopyCommitter(null, taskAttemptContext); fs = FileSystem.get(conf); FsPermission sourcePerm = new FsPermission((short) 511); FsPermission initialPerm = new FsPermission((short) 448); sourceBase = TestDistCpUtils.createTestSetup(fs, sourcePerm); workBase = TestDistCpUtils.createTestSetup(fs, initialPerm); targetBase = "/tmp1/" + rand.nextLong(); final DistCpOptions options = new DistCpOptions.Builder( Collections.singletonList(new Path(sourceBase)), new Path("/out")) .preserve(FileAttribute.PERMISSION).build(); options.appendToConf(conf); final DistCpContext context = new DistCpContext(options); context.setTargetPathExists(false); CopyListing listing = new GlobbedCopyListing(conf, CREDENTIALS); Path listingFile = new Path("/tmp1/" + rand.nextLong()); listing.buildListing(listingFile, context); conf.set(CONF_LABEL_TARGET_FINAL_PATH, targetBase); conf.set(CONF_LABEL_TARGET_WORK_PATH, workBase); conf.setBoolean(DistCpConstants.CONF_LABEL_ATOMIC_COPY, true); committer.commitJob(jobContext); checkDirectoryPermissions(fs, targetBase, sourcePerm); } finally { TestDistCpUtils.delete(fs, "/tmp1"); conf.unset(DistCpConstants.CONF_LABEL_PRESERVE_STATUS); } }
@Test public void testDeleteMissingFlatInterleavedFiles() 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 = "/tmp1/" + String.valueOf(rand.nextLong()); targetBase = "/tmp1/" + String.valueOf(rand.nextLong()); createFile(fs, sourceBase + "/1"); createFile(fs, sourceBase + "/3"); createFile(fs, sourceBase + "/4"); createFile(fs, sourceBase + "/5"); createFile(fs, sourceBase + "/7"); createFile(fs, sourceBase + "/8"); createFile(fs, sourceBase + "/9"); createFile(fs, targetBase + "/2"); createFile(fs, targetBase + "/4"); createFile(fs, targetBase + "/5"); createFile(fs, targetBase + "/7"); createFile(fs, targetBase + "/9"); createFile(fs, targetBase + "/A"); 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); Assert.assertEquals(4, fs.listStatus(new Path(targetBase)).length); committer.commitJob(jobContext); verifyFoldersAreInSync(fs, targetBase, sourceBase); Assert.assertEquals(4, fs.listStatus(new Path(targetBase)).length); } finally { TestDistCpUtils.delete(fs, "/tmp1"); conf.set(DistCpConstants.CONF_LABEL_DELETE_MISSING, "false"); } }
@Test public void testAtomicCommitMissingFinal() 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)); 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)); assertPathDoesNotExist(fs, "Final path", new Path(finalPath)); committer.commitJob(jobContext); assertPathDoesNotExist(fs, "Work path", new Path(workPath)); assertPathExists(fs, "Final path", new Path(finalPath)); committer.commitJob(jobContext); assertPathDoesNotExist(fs, "Work path", new Path(workPath)); assertPathExists(fs, "Final path", new Path(finalPath)); } finally { TestDistCpUtils.delete(fs, workPath); TestDistCpUtils.delete(fs, finalPath); conf.setBoolean(DistCpConstants.CONF_LABEL_ATOMIC_COPY, false); } }
@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); } }