DistCpProcedure extends BalanceProcedure { void diffDistCp() throws IOException, RetryException { RunningJobStatus job = getCurrentJob(); if (job != null) { if (job.isComplete()) { jobId = null; if (job.isSuccessful()) { LOG.info("DistCp succeeded. jobId={}", job.getJobID()); } else { throw new IOException("DistCp failed. jobId=" + job.getJobID() + " failure=" + job.getFailureInfo()); } } else { throw new RetryException(); } } else if (!verifyDiff()) { if (!verifyOpenFiles() || forceCloseOpenFiles) { updateStage(Stage.DISABLE_WRITE); } else { throw new RetryException(); } } else { submitDiffDistCp(); } } DistCpProcedure();  DistCpProcedure(String name, String nextProcedure, long delayDuration,
      FedBalanceContext context); @Override boolean execute(); @Override void write(DataOutput out); @Override void readFields(DataInput in); static final Logger LOG; }
@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)); fs.rename(new Path(src, "a"), new Path("/a")); executeProcedure(dcProcedure, Stage.FINISH, () -> dcProcedure.finalDistCp()); assertFalse(fs.exists(new Path(dst, "a"))); fs.rename(new Path("/a"), new Path(src, "a")); executeProcedure(dcProcedure, Stage.FINISH, () -> dcProcedure.finalDistCp()); assertTrue(fs.exists(new Path(dst, "a"))); 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)); }