void initDistCp() throws IOException, RetryException {
    RunningJobStatus job = getCurrentJob();
    if (job != null) {
      // the distcp has been submitted.
      if (job.isComplete()) {
        jobId = null; // unset jobId because the job is done.
        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);
    }
  }
@Test
  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);
    // set permission.
    fs.setPermission(src, FsPermission.createImmutable((short) 020));

    FedBalanceContext context = buildContext(src, dst, MOUNT);
    DistCpProcedure dcProcedure =
        new DistCpProcedure("distcp-procedure", null, 1000, context);

    // submit distcp.
    try {
      dcProcedure.initDistCp();
    } catch (RetryException e) {
    }
    fs.delete(new Path(src, "a"), true);
    // wait until job done.
    executeProcedure(dcProcedure, Stage.DIFF_DISTCP,
        () -> dcProcedure.initDistCp());
    assertTrue(fs.exists(dst));
    // Because we used snapshot, the file should be copied.
    assertTrue(fs.exists(new Path(dst, "a")));
    cleanup(fs, new Path(testRoot));
  }