public boolean sync() throws IOException {
    if (!preSyncCheck()) {
      return false;
    }

    if (!getAllDiffs()) {
      return false;
    }

    List<Path> sourcePaths = context.getSourcePaths();
    final Path sourceDir = sourcePaths.get(0);
    final Path targetDir = context.getTargetPath();
    final FileSystem tfs = targetDir.getFileSystem(conf);

    Path tmpDir = null;
    try {
      tmpDir = createTargetTmpDir(tfs, targetDir);
      DiffInfo[] renameAndDeleteDiffs =
          getRenameAndDeleteDiffsForSync(targetDir);
      if (renameAndDeleteDiffs.length > 0) {
        // do the real sync work: deletion and rename
        syncDiff(renameAndDeleteDiffs, tfs, tmpDir);
      }
      return true;
    } catch (Exception e) {
      DistCp.LOG.warn("Failed to use snapshot diff for distcp", e);
      return false;
    } finally {
      deleteTargetTmpDir(tfs, tmpDir);
      // TODO: since we have tmp directory, we can support "undo" with failures
      // set the source path using the snapshot path
      context.setSourcePaths(Arrays.asList(getSnapshotPath(sourceDir,
          context.getToSnapshot())));
    }
  }
@Test
  public void testSync() throws Exception {
    initData(source);
    initData(target);
    enableAndCreateFirstSnapshot();

    // make changes under source
    int numCreatedModified = changeData(dfs, source);
    dfs.createSnapshot(source, "s2");

    // before sync, make some further changes on source. this should not affect
    // the later distcp since we're copying (s2-s1) to target
    final Path toDelete = new Path(source, "foo/d1/foo/f1");
    dfs.delete(toDelete, true);
    final Path newdir = new Path(source, "foo/d1/foo/newdir");
    dfs.mkdirs(newdir);

    SnapshotDiffReport report = dfs.getSnapshotDiffReport(source, "s1", "s2");
    System.out.println(report);

    DistCpSync distCpSync = new DistCpSync(context, conf);

    // do the sync
    Assert.assertTrue(distCpSync.sync());

    // make sure the source path has been updated to the snapshot path
    final Path spath = new Path(source,
            HdfsConstants.DOT_SNAPSHOT_DIR + Path.SEPARATOR + "s2");
    Assert.assertEquals(spath, context.getSourcePaths().get(0));

    // build copy listing
    final Path listingPath = new Path("/tmp/META/fileList.seq");
    CopyListing listing = new SimpleCopyListing(conf, new Credentials(), distCpSync);
    listing.buildListing(listingPath, context);

    Map<Text, CopyListingFileStatus> copyListing = getListing(listingPath);
    CopyMapper copyMapper = new CopyMapper();
    StubContext stubContext = new StubContext(conf, null, 0);
    Mapper<Text, CopyListingFileStatus, Text, Text>.Context mapContext =
        stubContext.getContext();
    // Enable append
    mapContext.getConfiguration().setBoolean(
        DistCpOptionSwitch.APPEND.getConfigLabel(), true);
    copyMapper.setup(mapContext);
    for (Map.Entry<Text, CopyListingFileStatus> entry : copyListing.entrySet()) {
      copyMapper.map(entry.getKey(), entry.getValue(), mapContext);
    }

    // verify that we only list modified and created files/directories
    Assert.assertEquals(numCreatedModified, copyListing.size());

    // verify that we only copied new appended data of f2 and the new file f1
    Assert.assertEquals(BLOCK_SIZE * 3, stubContext.getReporter()
        .getCounter(CopyMapper.Counter.BYTESCOPIED).getValue());

    // verify the source and target now has the same structure
    verifyCopy(dfs.getFileStatus(spath), dfs.getFileStatus(target), false);
  }