@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);
    //Test for idempotent commit
    committer.commitJob(jobContext);
    checkDirectoryPermissions(fs, targetBase, sourcePerm);
  } finally {
    TestDistCpUtils.delete(fs, "/tmp1");
    conf.unset(DistCpConstants.CONF_LABEL_PRESERVE_STATUS);
  }
}