CopyOutputFormat extends TextOutputFormat<K, V> { @Override public void checkOutputSpecs(JobContext context) throws IOException { Configuration conf = HadoopCompat.getConfiguration(context); if (getCommitDirectory(conf) == null) { throw new IllegalStateException("Commit directory not configured"); } Path workingPath = getWorkingDirectory(conf); if (workingPath == null) { throw new IllegalStateException("Working directory not configured"); } TokenCache.obtainTokensForNamenodes(HadoopCompat.getCredentials(context), new Path[] {workingPath}, conf); }  static void setWorkingDirectory(Job job, Path workingDirectory); static void setCommitDirectory(Job job, Path commitDirectory); static Path getWorkingDirectory(Job job); static Path getCommitDirectory(Job job); @Override OutputCommitter getOutputCommitter(TaskAttemptContext context); @Override void checkOutputSpecs(JobContext context);  }
@Test public void testCheckOutputSpecs() { try { OutputFormat outputFormat = new CopyOutputFormat(); Configuration conf = new Configuration(); Job job = new Job(conf); JobID jobID = new JobID("200707121733", 1); try { JobContext context = Mockito.mock(JobContext.class); Mockito.when(context.getConfiguration()).thenReturn(job.getConfiguration()); Mockito.when(context.getJobID()).thenReturn(jobID); outputFormat.checkOutputSpecs(context); Assert.fail("No checking for invalid work/commit path"); } catch (IllegalStateException ignore) { } CopyOutputFormat.setWorkingDirectory(job, new Path("/tmp/work")); try { JobContext context = Mockito.mock(JobContext.class); Mockito.when(context.getConfiguration()).thenReturn(job.getConfiguration()); Mockito.when(context.getJobID()).thenReturn(jobID); outputFormat.checkOutputSpecs(context); Assert.fail("No checking for invalid commit path"); } catch (IllegalStateException ignore) { } job.getConfiguration().set(DistCpConstants.CONF_LABEL_TARGET_WORK_PATH, ""); CopyOutputFormat.setCommitDirectory(job, new Path("/tmp/commit")); try { JobContext context = Mockito.mock(JobContext.class); Mockito.when(context.getConfiguration()).thenReturn(job.getConfiguration()); Mockito.when(context.getJobID()).thenReturn(jobID); outputFormat.checkOutputSpecs(context); Assert.fail("No checking for invalid work path"); } catch (IllegalStateException ignore) { } CopyOutputFormat.setWorkingDirectory(job, new Path("/tmp/work")); CopyOutputFormat.setCommitDirectory(job, new Path("/tmp/commit")); try { JobContext context = Mockito.mock(JobContext.class); Mockito.when(context.getConfiguration()).thenReturn(job.getConfiguration()); Mockito.when(context.getJobID()).thenReturn(jobID); outputFormat.checkOutputSpecs(context); } catch (IllegalStateException ignore) { ignore.printStackTrace(); Assert.fail("Output spec check failed."); } } catch (IOException e) { LOG.error("Exception encountered while testing checkoutput specs", e); Assert.fail("Checkoutput Spec failure"); } catch (InterruptedException e) { LOG.error("Exception encountered while testing checkoutput specs", e); Assert.fail("Checkoutput Spec failure"); } }