DistCpProcedure extends BalanceProcedure { void initDistCp() throws IOException, RetryException { RunningJobStatus job = getCurrentJob(); if (job != null) { if (job.isComplete()) { jobId = null; if (job.isSuccessful()) { updateStage(Stage.DIFF_DISTCP); return; } else { LOG.warn("DistCp failed. Failure={}", job.getFailureInfo()); } } else { throw new RetryException(); } } else { pathCheckBeforeInitDistcp(); srcFs.createSnapshot(src, CURRENT_SNAPSHOT_NAME); jobId = submitDistCpJob( src.toString() + HdfsConstants.SEPARATOR_DOT_SNAPSHOT_DIR_SEPARATOR + CURRENT_SNAPSHOT_NAME, dst.toString(), false); } } 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 testInitDistCp() 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); fs.setPermission(src, FsPermission.createImmutable((short) 020)); FedBalanceContext context = buildContext(src, dst, MOUNT); DistCpProcedure dcProcedure = new DistCpProcedure("distcp-procedure", null, 1000, context); try { dcProcedure.initDistCp(); } catch (RetryException e) { } fs.delete(new Path(src, "a"), true); executeProcedure(dcProcedure, Stage.DIFF_DISTCP, () -> dcProcedure.initDistCp()); assertTrue(fs.exists(dst)); assertTrue(fs.exists(new Path(dst, "a"))); cleanup(fs, new Path(testRoot)); }