@Test() public void testAbortJob() throws IOException { for (Path stagingDir : this.stagingDirs) { Assert.assertTrue(this.fs.exists(stagingDir)); } Configuration conf = new Configuration(); conf.set(ConfigurationKeys.FS_URI_KEY, ConfigurationKeys.LOCAL_FS_URI); conf.set(ConfigurationKeys.MR_JOB_ROOT_DIR_KEY, OUTPUT_PATH.toString()); conf.set(ConfigurationKeys.JOB_NAME_KEY, JOB_NAME); new GobblinOutputCommitter().abortJob(Job.getInstance(conf), JobStatus.State.RUNNING); for (Path stagingDir : this.stagingDirs) { Assert.assertTrue(!this.fs.exists(stagingDir)); } }
@Override public void abortJob(JobContext jobContext, JobStatus.State state) throws IOException { LOG.info("Aborting Job: " + jobContext.getJobID() + " with state: " + state); Configuration conf = jobContext.getConfiguration(); URI fsUri = URI.create(conf.get(ConfigurationKeys.FS_URI_KEY, ConfigurationKeys.LOCAL_FS_URI)); FileSystem fs = FileSystem.get(fsUri, conf); Path mrJobDir = new Path(conf.get(ConfigurationKeys.MR_JOB_ROOT_DIR_KEY), conf.get(ConfigurationKeys.JOB_NAME_KEY)); Path jobInputDir = new Path(mrJobDir, MRJobLauncher.INPUT_DIR_NAME); if (!fs.exists(jobInputDir) || !fs.isDirectory(jobInputDir)) { LOG.warn(String.format("%s either does not exist or is not a directory. No data to cleanup.", jobInputDir)); return; } try { for (FileStatus status : fs.listStatus(jobInputDir, new WorkUnitFilter())) { Closer workUnitFileCloser = Closer.create(); if (status.getPath().getName().endsWith(AbstractJobLauncher.WORK_UNIT_FILE_EXTENSION)) { WorkUnit wu = WorkUnit.createEmpty(); try { wu.readFields(workUnitFileCloser.register(new DataInputStream(fs.open(status.getPath())))); } finally { workUnitFileCloser.close(); } JobLauncherUtils.cleanTaskStagingData(new WorkUnitState(wu), LOG); } if (status.getPath().getName().endsWith(AbstractJobLauncher.MULTI_WORK_UNIT_FILE_EXTENSION)) { MultiWorkUnit mwu = MultiWorkUnit.createEmpty(); try { mwu.readFields(workUnitFileCloser.register(new DataInputStream(fs.open(status.getPath())))); } finally { workUnitFileCloser.close(); } for (WorkUnit wu : mwu.getWorkUnits()) { JobLauncherUtils.cleanTaskStagingData(new WorkUnitState(wu), LOG); } } } } finally { try { cleanUpWorkingDirectory(mrJobDir, fs); } finally { super.abortJob(jobContext, state); } } }
GobblinOutputCommitter extends OutputCommitter { @Override public void abortJob(JobContext jobContext, JobStatus.State state) throws IOException { LOG.info("Aborting Job: " + jobContext.getJobID() + " with state: " + state); Configuration conf = jobContext.getConfiguration(); URI fsUri = URI.create(conf.get(ConfigurationKeys.FS_URI_KEY, ConfigurationKeys.LOCAL_FS_URI)); FileSystem fs = FileSystem.get(fsUri, conf); Path mrJobDir = new Path(conf.get(ConfigurationKeys.MR_JOB_ROOT_DIR_KEY), conf.get(ConfigurationKeys.JOB_NAME_KEY)); Path jobInputDir = new Path(mrJobDir, MRJobLauncher.INPUT_DIR_NAME); if (!fs.exists(jobInputDir) || !fs.isDirectory(jobInputDir)) { LOG.warn(String.format("%s either does not exist or is not a directory. No data to cleanup.", jobInputDir)); return; } try { for (FileStatus status : fs.listStatus(jobInputDir, new WorkUnitFilter())) { Closer workUnitFileCloser = Closer.create(); if (status.getPath().getName().endsWith(AbstractJobLauncher.WORK_UNIT_FILE_EXTENSION)) { WorkUnit wu = WorkUnit.createEmpty(); try { wu.readFields(workUnitFileCloser.register(new DataInputStream(fs.open(status.getPath())))); } finally { workUnitFileCloser.close(); } JobLauncherUtils.cleanTaskStagingData(new WorkUnitState(wu), LOG); } if (status.getPath().getName().endsWith(AbstractJobLauncher.MULTI_WORK_UNIT_FILE_EXTENSION)) { MultiWorkUnit mwu = MultiWorkUnit.createEmpty(); try { mwu.readFields(workUnitFileCloser.register(new DataInputStream(fs.open(status.getPath())))); } finally { workUnitFileCloser.close(); } for (WorkUnit wu : mwu.getWorkUnits()) { JobLauncherUtils.cleanTaskStagingData(new WorkUnitState(wu), LOG); } } } } finally { try { cleanUpWorkingDirectory(mrJobDir, fs); } finally { super.abortJob(jobContext, state); } } } }
GobblinOutputCommitter extends OutputCommitter { @Override public void abortJob(JobContext jobContext, JobStatus.State state) throws IOException { LOG.info("Aborting Job: " + jobContext.getJobID() + " with state: " + state); Configuration conf = jobContext.getConfiguration(); URI fsUri = URI.create(conf.get(ConfigurationKeys.FS_URI_KEY, ConfigurationKeys.LOCAL_FS_URI)); FileSystem fs = FileSystem.get(fsUri, conf); Path mrJobDir = new Path(conf.get(ConfigurationKeys.MR_JOB_ROOT_DIR_KEY), conf.get(ConfigurationKeys.JOB_NAME_KEY)); Path jobInputDir = new Path(mrJobDir, MRJobLauncher.INPUT_DIR_NAME); if (!fs.exists(jobInputDir) || !fs.isDirectory(jobInputDir)) { LOG.warn(String.format("%s either does not exist or is not a directory. No data to cleanup.", jobInputDir)); return; } try { for (FileStatus status : fs.listStatus(jobInputDir, new WorkUnitFilter())) { Closer workUnitFileCloser = Closer.create(); if (status.getPath().getName().endsWith(AbstractJobLauncher.WORK_UNIT_FILE_EXTENSION)) { WorkUnit wu = WorkUnit.createEmpty(); try { wu.readFields(workUnitFileCloser.register(new DataInputStream(fs.open(status.getPath())))); } finally { workUnitFileCloser.close(); } JobLauncherUtils.cleanTaskStagingData(new WorkUnitState(wu), LOG); } if (status.getPath().getName().endsWith(AbstractJobLauncher.MULTI_WORK_UNIT_FILE_EXTENSION)) { MultiWorkUnit mwu = MultiWorkUnit.createEmpty(); try { mwu.readFields(workUnitFileCloser.register(new DataInputStream(fs.open(status.getPath())))); } finally { workUnitFileCloser.close(); } for (WorkUnit wu : mwu.getWorkUnits()) { JobLauncherUtils.cleanTaskStagingData(new WorkUnitState(wu), LOG); } } } } finally { try { cleanUpWorkingDirectory(mrJobDir, fs); } finally { super.abortJob(jobContext, state); } } }  }
GobblinOutputCommitter extends OutputCommitter { @Override public void abortJob(JobContext jobContext, JobStatus.State state) throws IOException { LOG.info("Aborting Job: " + jobContext.getJobID() + " with state: " + state); Configuration conf = jobContext.getConfiguration(); URI fsUri = URI.create(conf.get(ConfigurationKeys.FS_URI_KEY, ConfigurationKeys.LOCAL_FS_URI)); FileSystem fs = FileSystem.get(fsUri, conf); Path mrJobDir = new Path(conf.get(ConfigurationKeys.MR_JOB_ROOT_DIR_KEY), conf.get(ConfigurationKeys.JOB_NAME_KEY)); Path jobInputDir = new Path(mrJobDir, MRJobLauncher.INPUT_DIR_NAME); if (!fs.exists(jobInputDir) || !fs.isDirectory(jobInputDir)) { LOG.warn(String.format("%s either does not exist or is not a directory. No data to cleanup.", jobInputDir)); return; } try { for (FileStatus status : fs.listStatus(jobInputDir, new WorkUnitFilter())) { Closer workUnitFileCloser = Closer.create(); if (status.getPath().getName().endsWith(AbstractJobLauncher.WORK_UNIT_FILE_EXTENSION)) { WorkUnit wu = WorkUnit.createEmpty(); try { wu.readFields(workUnitFileCloser.register(new DataInputStream(fs.open(status.getPath())))); } finally { workUnitFileCloser.close(); } JobLauncherUtils.cleanTaskStagingData(new WorkUnitState(wu), LOG); } if (status.getPath().getName().endsWith(AbstractJobLauncher.MULTI_WORK_UNIT_FILE_EXTENSION)) { MultiWorkUnit mwu = MultiWorkUnit.createEmpty(); try { mwu.readFields(workUnitFileCloser.register(new DataInputStream(fs.open(status.getPath())))); } finally { workUnitFileCloser.close(); } for (WorkUnit wu : mwu.getWorkUnits()) { JobLauncherUtils.cleanTaskStagingData(new WorkUnitState(wu), LOG); } } } } finally { try { cleanUpWorkingDirectory(mrJobDir, fs); } finally { super.abortJob(jobContext, state); } } }  @Override void abortJob(JobContext jobContext, JobStatus.State state); @Override void abortTask(TaskAttemptContext arg0); @Override void commitTask(TaskAttemptContext arg0); @Override boolean needsTaskCommit(TaskAttemptContext arg0); @Override void setupJob(JobContext arg0); @Override void setupTask(TaskAttemptContext arg0); @Override boolean isRecoverySupported(); @Override void recoverTask(TaskAttemptContext taskContext); }
GobblinOutputCommitter extends OutputCommitter { @Override public void abortJob(JobContext jobContext, JobStatus.State state) throws IOException { LOG.info("Aborting Job: " + jobContext.getJobID() + " with state: " + state); Configuration conf = jobContext.getConfiguration(); URI fsUri = URI.create(conf.get(ConfigurationKeys.FS_URI_KEY, ConfigurationKeys.LOCAL_FS_URI)); FileSystem fs = FileSystem.get(fsUri, conf); Path mrJobDir = new Path(conf.get(ConfigurationKeys.MR_JOB_ROOT_DIR_KEY), conf.get(ConfigurationKeys.JOB_NAME_KEY)); Path jobInputDir = new Path(mrJobDir, MRJobLauncher.INPUT_DIR_NAME); if (!fs.exists(jobInputDir) || !fs.isDirectory(jobInputDir)) { LOG.warn(String.format("%s either does not exist or is not a directory. No data to cleanup.", jobInputDir)); return; } try { for (FileStatus status : fs.listStatus(jobInputDir, new WorkUnitFilter())) { Closer workUnitFileCloser = Closer.create(); if (status.getPath().getName().endsWith(AbstractJobLauncher.WORK_UNIT_FILE_EXTENSION)) { WorkUnit wu = WorkUnit.createEmpty(); try { wu.readFields(workUnitFileCloser.register(new DataInputStream(fs.open(status.getPath())))); } finally { workUnitFileCloser.close(); } JobLauncherUtils.cleanTaskStagingData(new WorkUnitState(wu), LOG); } if (status.getPath().getName().endsWith(AbstractJobLauncher.MULTI_WORK_UNIT_FILE_EXTENSION)) { MultiWorkUnit mwu = MultiWorkUnit.createEmpty(); try { mwu.readFields(workUnitFileCloser.register(new DataInputStream(fs.open(status.getPath())))); } finally { workUnitFileCloser.close(); } for (WorkUnit wu : mwu.getWorkUnits()) { JobLauncherUtils.cleanTaskStagingData(new WorkUnitState(wu), LOG); } } } } finally { try { cleanUpWorkingDirectory(mrJobDir, fs); } finally { super.abortJob(jobContext, state); } } }  @Override void abortJob(JobContext jobContext, JobStatus.State state); @Override void abortTask(TaskAttemptContext arg0); @Override void commitTask(TaskAttemptContext arg0); @Override boolean needsTaskCommit(TaskAttemptContext arg0); @Override void setupJob(JobContext arg0); @Override void setupTask(TaskAttemptContext arg0); @Override boolean isRecoverySupported(); @Override void recoverTask(TaskAttemptContext taskContext);  }