@Override
  public void checkOutputSpecs(JobContext context) throws IOException {
    Configuration conf = context.getConfiguration();

    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");
    }

    // get delegation token for outDir's file system
    TokenCache.obtainTokensForNamenodes(context.getCredentials(),
                                        new Path[] {workingPath}, conf);
  }
@Test
  public void testCheckOutputSpecs() {
    try {
      OutputFormat outputFormat = new CopyOutputFormat();
      Job job = Job.getInstance(new Configuration());
      JobID jobID = new JobID("200707121733", 1);

      try {
        JobContext context = new JobContextImpl(job.getConfiguration(), 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 = new JobContextImpl(job.getConfiguration(), 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 = new JobContextImpl(job.getConfiguration(), 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 = new JobContextImpl(job.getConfiguration(), jobID);
        outputFormat.checkOutputSpecs(context);
      } catch (IllegalStateException ignore) {
        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");
    }
  }