DistCpSync { 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); final DistributedFileSystem targetFs = (DistributedFileSystem) tfs; Path tmpDir = null; try { tmpDir = createTargetTmpDir(targetFs, targetDir); DiffInfo[] renameAndDeleteDiffs = getRenameAndDeleteDiffsForSync(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); context.setSourcePaths(Arrays.asList(getSnapshotPath(sourceDir, context.getToSnapshot()))); } } DistCpSync(DistCpContext context, Configuration conf); boolean sync(); ArrayList<DiffInfo> prepareDiffListForCopyListing(); 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, context.getSourcePaths().get(0)); context.setSourcePaths(Collections.singletonList(source)); dfs.allowSnapshot(source); dfs.allowSnapshot(target); Assert.assertFalse(sync()); Assert.assertEquals(spath, context.getSourcePaths().get(0)); context.setSourcePaths(Collections.singletonList(source)); dfs.createSnapshot(source, "s1"); dfs.createSnapshot(source, "s2"); dfs.createSnapshot(target, "s1"); Assert.assertTrue(sync()); context.setSourcePaths(Collections.singletonList(source)); final Path subTarget = new Path(target, "sub"); dfs.mkdirs(subTarget); Assert.assertFalse(sync()); Assert.assertEquals(spath, context.getSourcePaths().get(0)); context.setSourcePaths(Collections.singletonList(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(context, conf); Assert.assertTrue(distCpSync.sync()); final Path spath = new Path(source, HdfsConstants.DOT_SNAPSHOT_DIR + Path.SEPARATOR + "s2"); Assert.assertEquals(spath, context.getSourcePaths().get(0)); 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(); 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); } 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 { final DistCpOptions options = new DistCpOptions.Builder( Collections.singletonList(source), target) .withSyncFolder(true) .withUseDiff("s1", ".") .build(); context = new DistCpContext(options); initData(source); initData(target); enableAndCreateFirstSnapshot(); changeData(source); sync(); Assert.assertEquals(source, context.getSourcePaths().get(0)); }
@Test public void testSyncSnapshotTimeStampChecking() throws Exception { initData(source); initData(target); dfs.allowSnapshot(source); dfs.allowSnapshot(target); dfs.createSnapshot(source, "s2"); dfs.createSnapshot(target, "s1"); Thread.sleep(1000); dfs.createSnapshot(source, "s1"); boolean threwException = false; try { DistCpSync distCpSync = new DistCpSync(context, conf); distCpSync.sync(); } catch (HadoopIllegalArgumentException e) { threwException = true; GenericTestUtils.assertExceptionContains( "Snapshot s2 should be newer than s1", e); } Assert.assertTrue(threwException); }