DistCpSync { public boolean sync() throws IOException { if (!preSyncCheck()) { return false; } if (!getAllDiffs()) { return false; } List<Path> sourcePaths = inputOptions.getSourcePaths(); final Path sourceDir = sourcePaths.get(0); final Path targetDir = inputOptions.getTargetPath(); final FileSystem tfs = targetDir.getFileSystem(conf); final DistributedFileSystem targetFs = (DistributedFileSystem) tfs; Path tmpDir = null; try { tmpDir = createTargetTmpDir(targetFs, targetDir); DiffInfo[] renameAndDeleteDiffs = getRenameAndDeleteDiffs(targetDir); if (renameAndDeleteDiffs.length > 0) { syncDiff(renameAndDeleteDiffs, targetFs, tmpDir); } return true; } catch (Exception e) { DistCp.LOG.warn("Failed to use snapshot diff for distcp", e); return false; } finally { deleteTargetTmpDir(targetFs, tmpDir); inputOptions.setSourcePaths(Arrays.asList(getSourceSnapshotPath(sourceDir, inputOptions.getToSnapshot()))); } } DistCpSync(DistCpOptions options, Configuration conf); boolean sync(); ArrayList<DiffInfo> prepareDiffList(); HashSet<String> getTraverseExcludeList(Path newDir, Path prefix);  }
@Test public void testFallback() throws Exception { Assert.assertFalse(sync()); final Path spath = new Path(source, HdfsConstants.DOT_SNAPSHOT_DIR + Path.SEPARATOR + "s2"); Assert.assertEquals(spath, options.getSourcePaths().get(0)); options.setSourcePaths(Arrays.asList(source)); dfs.allowSnapshot(source); dfs.allowSnapshot(target); Assert.assertFalse(sync()); Assert.assertEquals(spath, options.getSourcePaths().get(0)); options.setSourcePaths(Arrays.asList(source)); dfs.createSnapshot(source, "s1"); dfs.createSnapshot(source, "s2"); dfs.createSnapshot(target, "s1"); Assert.assertTrue(sync()); options.setSourcePaths(Arrays.asList(source)); final Path subTarget = new Path(target, "sub"); dfs.mkdirs(subTarget); Assert.assertFalse(sync()); Assert.assertEquals(spath, options.getSourcePaths().get(0)); options.setSourcePaths(Arrays.asList(source)); dfs.delete(subTarget, true); Assert.assertTrue(sync()); }
@Test public void testSync() throws Exception { initData(source); initData(target); enableAndCreateFirstSnapshot(); int numCreatedModified = changeData(source); dfs.createSnapshot(source, "s2"); 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(options, conf); Assert.assertTrue(distCpSync.sync()); final Path spath = new Path(source, HdfsConstants.DOT_SNAPSHOT_DIR + Path.SEPARATOR + "s2"); Assert.assertEquals(spath, options.getSourcePaths().get(0)); final Path listingPath = new Path("/tmp/META/fileList.seq"); CopyListing listing = new SimpleCopyListing(conf, new Credentials(), distCpSync); listing.buildListing(listingPath, options); Map<Text, CopyListingFileStatus> copyListing = getListing(listingPath); CopyMapper copyMapper = new CopyMapper(); StubContext stubContext = new StubContext(conf, null, 0); Mapper<Text, CopyListingFileStatus, Text, Text>.Context context = stubContext.getContext(); context.getConfiguration().setBoolean( DistCpOptionSwitch.APPEND.getConfigLabel(), true); copyMapper.setup(context); for (Map.Entry<Text, CopyListingFileStatus> entry : copyListing.entrySet()) { copyMapper.map(entry.getKey(), entry.getValue(), context); } Assert.assertEquals(numCreatedModified, copyListing.size()); Assert.assertEquals(BLOCK_SIZE * 3, stubContext.getReporter() .getCounter(CopyMapper.Counter.BYTESCOPIED).getValue()); verifyCopy(dfs.getFileStatus(spath), dfs.getFileStatus(target), false); }
@Test public void testSyncWithCurrent() throws Exception { options.setUseDiff(true, "s1", "."); initData(source); initData(target); enableAndCreateFirstSnapshot(); changeData(source); sync(); Assert.assertEquals(source, options.getSourcePaths().get(0)); }