StagingCommitter extends AbstractS3ACommitter { public static Path getJobAttemptPath(JobContext context, Path out) { return getJobAttemptPath(getAppAttemptId(context), out); } StagingCommitter(Path outputPath,
      TaskAttemptContext context); @Override String getName(); @Override String toString(); static String getUploadUUID(Configuration conf, String jobId); static String getUploadUUID(Configuration conf, JobID jobId); Boolean useUniqueFilenames(); FileSystem getJobAttemptFileSystem(JobContext context); static Path getJobAttemptPath(JobContext context, Path out); static Path getTaskAttemptPath(TaskAttemptContext context, Path out); Path getCommittedTaskPath(TaskAttemptContext context); @Override Path getTempTaskAttemptPath(TaskAttemptContext context); @Override Path getBaseTaskAttemptPath(TaskAttemptContext context); @Override Path getJobAttemptPath(JobContext context); @Override void setupJob(JobContext context); @Override void cleanupStagingDirs(); @Override void setupTask(TaskAttemptContext context); @Override boolean needsTaskCommit(TaskAttemptContext context); @Override void commitTask(TaskAttemptContext context); @Override void abortTask(TaskAttemptContext context); final ConflictResolution getConflictResolutionMode(
      JobContext context,
      Configuration fsConf); static String getConfictModeOption(JobContext context,
      Configuration fsConf, String defVal); @Override void preCommitJob(
      final JobContext context,
      final ActiveCommit pending); static final String NAME; }
@Test public void testJobCommit() throws Exception { Path jobAttemptPath = jobCommitter.getJobAttemptPath(job); FileSystem fs = jobAttemptPath.getFileSystem(conf); Set<String> uploads = runTasks(job, 4, 3); assertNotEquals(0, uploads.size()); assertPathExists(fs, "No job attempt path", jobAttemptPath); jobCommitter.commitJob(job); assertEquals("Should have aborted no uploads", 0, results.getAborts().size()); assertEquals("Should have deleted no uploads", 0, results.getDeletes().size()); assertEquals("Should have committed all uploads", uploads, getCommittedIds(results.getCommits())); assertPathDoesNotExist(fs, "jobAttemptPath not deleted", jobAttemptPath); }
@Test public void testJobAbort() throws Exception { Path jobAttemptPath = jobCommitter.getJobAttemptPath(job); FileSystem fs = jobAttemptPath.getFileSystem(conf); Set<String> uploads = runTasks(job, 4, 3); assertPathExists(fs, "No job attempt path", jobAttemptPath); jobCommitter.abortJob(job, JobStatus.State.KILLED); assertEquals("Should have committed no uploads: " + jobCommitter, 0, results.getCommits().size()); assertEquals("Should have deleted no uploads: " + jobCommitter, 0, results.getDeletes().size()); assertEquals("Should have aborted all uploads: " + jobCommitter, uploads, getAbortedIds(results.getAborts())); assertPathDoesNotExist(fs, "jobAttemptPath not deleted", jobAttemptPath); }