@Test(timeout = 30000)
public void testDiffDistCp() throws Exception {

  String testRoot = nnUri + "/user/foo/testdir." + getMethodName();
  DistributedFileSystem fs =
      (DistributedFileSystem) FileSystem.get(URI.create(nnUri), conf);
  createFiles(fs, testRoot, srcfiles);
  Path src = new Path(testRoot, SRCDAT);
  Path dst = new Path(testRoot, DSTDAT);
  FedBalanceContext context = buildContext(src, dst, MOUNT);
  DistCpProcedure dcProcedure =
      new DistCpProcedure("distcp-procedure", null, 1000, context);
  executeProcedure(dcProcedure, Stage.DIFF_DISTCP,
      () -> dcProcedure.initDistCp());
  assertTrue(fs.exists(dst));
  // move file out of src and test distcp.
  fs.rename(new Path(src, "a"), new Path("/a"));
  executeProcedure(dcProcedure, Stage.FINISH,
      () -> dcProcedure.finalDistCp());
  assertFalse(fs.exists(new Path(dst, "a")));
  // move back file src/a and test distcp.
  fs.rename(new Path("/a"), new Path(src, "a"));
  executeProcedure(dcProcedure, Stage.FINISH,
      () -> dcProcedure.finalDistCp());
  assertTrue(fs.exists(new Path(dst, "a")));
  // append file src/a and test.
  OutputStream out = fs.append(new Path(src, "a"));
  out.write("hello".getBytes());
  out.close();
  long len = fs.getFileStatus(new Path(src, "a")).getLen();
  executeProcedure(dcProcedure, Stage.FINISH,
      () -> dcProcedure.finalDistCp());
  assertEquals(len, fs.getFileStatus(new Path(dst, "a")).getLen());
  cleanup(fs, new Path(testRoot));
}