@Test public void testSync() throws Exception { initData(source); initData(target); dfs.allowSnapshot(source); dfs.allowSnapshot(target); dfs.createSnapshot(source, "s1"); dfs.createSnapshot(target, "s1"); 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); Assert.assertTrue(DistCpSync.sync(options, conf)); 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 GlobbedCopyListing(conf, new Credentials()); 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(BLOCK_SIZE * 3, stubContext.getReporter() .getCounter(CopyMapper.Counter.BYTESCOPIED).getValue()); verifyCopy(dfs.getFileStatus(spath), dfs.getFileStatus(target), false); }
static boolean sync(DistCpOptions inputOptions, Configuration conf) throws IOException { List<Path> sourcePaths = inputOptions.getSourcePaths(); if (sourcePaths.size() != 1) { throw new IllegalArgumentException(sourcePaths.size() + " source paths are provided"); } final Path sourceDir = sourcePaths.get(0); final Path targetDir = inputOptions.getTargetPath(); final FileSystem sfs = sourceDir.getFileSystem(conf); final FileSystem tfs = targetDir.getFileSystem(conf); if (!(sfs instanceof DistributedFileSystem) || !(tfs instanceof DistributedFileSystem)) { throw new IllegalArgumentException("The FileSystems needs to" + " be DistributedFileSystem for using snapshot-diff-based distcp"); } final DistributedFileSystem sourceFs = (DistributedFileSystem) sfs; final DistributedFileSystem targetFs= (DistributedFileSystem) tfs; if (!checkNoChange(inputOptions, targetFs, targetDir)) { inputOptions.setSourcePaths(Arrays.asList(getSourceSnapshotPath(sourceDir, inputOptions.getToSnapshot()))); return false; } Path tmpDir = null; try { tmpDir = createTargetTmpDir(targetFs, targetDir); DiffInfo[] diffs = getDiffs(inputOptions, sourceFs, sourceDir, targetDir); if (diffs == null) { return false; } syncDiff(diffs, 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 { static boolean sync(DistCpOptions inputOptions, Configuration conf) throws IOException { List<Path> sourcePaths = inputOptions.getSourcePaths(); if (sourcePaths.size() != 1) { throw new IllegalArgumentException(sourcePaths.size() + " source paths are provided"); } final Path sourceDir = sourcePaths.get(0); final Path targetDir = inputOptions.getTargetPath(); final FileSystem sfs = sourceDir.getFileSystem(conf); final FileSystem tfs = targetDir.getFileSystem(conf); if (!(sfs instanceof DistributedFileSystem) || !(tfs instanceof DistributedFileSystem)) { throw new IllegalArgumentException("The FileSystems needs to" + " be DistributedFileSystem for using snapshot-diff-based distcp"); } final DistributedFileSystem sourceFs = (DistributedFileSystem) sfs; final DistributedFileSystem targetFs= (DistributedFileSystem) tfs; if (!checkNoChange(inputOptions, targetFs, targetDir)) { inputOptions.setSourcePaths(Arrays.asList(getSourceSnapshotPath(sourceDir, inputOptions.getToSnapshot()))); return false; } Path tmpDir = null; try { tmpDir = createTargetTmpDir(targetFs, targetDir); DiffInfo[] diffs = getDiffs(inputOptions, sourceFs, sourceDir, targetDir); if (diffs == null) { return false; } syncDiff(diffs, 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 { static boolean sync(DistCpOptions inputOptions, Configuration conf) throws IOException { List<Path> sourcePaths = inputOptions.getSourcePaths(); if (sourcePaths.size() != 1) { throw new IllegalArgumentException(sourcePaths.size() + " source paths are provided"); } final Path sourceDir = sourcePaths.get(0); final Path targetDir = inputOptions.getTargetPath(); final FileSystem sfs = sourceDir.getFileSystem(conf); final FileSystem tfs = targetDir.getFileSystem(conf); if (!(sfs instanceof DistributedFileSystem) || !(tfs instanceof DistributedFileSystem)) { throw new IllegalArgumentException("The FileSystems needs to" + " be DistributedFileSystem for using snapshot-diff-based distcp"); } final DistributedFileSystem sourceFs = (DistributedFileSystem) sfs; final DistributedFileSystem targetFs= (DistributedFileSystem) tfs; if (!checkNoChange(inputOptions, targetFs, targetDir)) { inputOptions.setSourcePaths(Arrays.asList(getSourceSnapshotPath(sourceDir, inputOptions.getToSnapshot()))); return false; } Path tmpDir = null; try { tmpDir = createTargetTmpDir(targetFs, targetDir); DiffInfo[] diffs = getDiffs(inputOptions, sourceFs, sourceDir, targetDir); if (diffs == null) { return false; } syncDiff(diffs, 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 { static boolean sync(DistCpOptions inputOptions, Configuration conf) throws IOException { List<Path> sourcePaths = inputOptions.getSourcePaths(); if (sourcePaths.size() != 1) { throw new IllegalArgumentException(sourcePaths.size() + " source paths are provided"); } final Path sourceDir = sourcePaths.get(0); final Path targetDir = inputOptions.getTargetPath(); final FileSystem sfs = sourceDir.getFileSystem(conf); final FileSystem tfs = targetDir.getFileSystem(conf); if (!(sfs instanceof DistributedFileSystem) || !(tfs instanceof DistributedFileSystem)) { throw new IllegalArgumentException("The FileSystems needs to" + " be DistributedFileSystem for using snapshot-diff-based distcp"); } final DistributedFileSystem sourceFs = (DistributedFileSystem) sfs; final DistributedFileSystem targetFs= (DistributedFileSystem) tfs; if (!checkNoChange(inputOptions, targetFs, targetDir)) { inputOptions.setSourcePaths(Arrays.asList(getSourceSnapshotPath(sourceDir, inputOptions.getToSnapshot()))); return false; } Path tmpDir = null; try { tmpDir = createTargetTmpDir(targetFs, targetDir); DiffInfo[] diffs = getDiffs(inputOptions, sourceFs, sourceDir, targetDir); if (diffs == null) { return false; } syncDiff(diffs, 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 { static boolean sync(DistCpOptions inputOptions, Configuration conf) throws IOException { List<Path> sourcePaths = inputOptions.getSourcePaths(); if (sourcePaths.size() != 1) { throw new IllegalArgumentException(sourcePaths.size() + " source paths are provided"); } final Path sourceDir = sourcePaths.get(0); final Path targetDir = inputOptions.getTargetPath(); final FileSystem sfs = sourceDir.getFileSystem(conf); final FileSystem tfs = targetDir.getFileSystem(conf); if (!(sfs instanceof DistributedFileSystem) || !(tfs instanceof DistributedFileSystem)) { throw new IllegalArgumentException("The FileSystems needs to" + " be DistributedFileSystem for using snapshot-diff-based distcp"); } final DistributedFileSystem sourceFs = (DistributedFileSystem) sfs; final DistributedFileSystem targetFs= (DistributedFileSystem) tfs; if (!checkNoChange(inputOptions, targetFs, targetDir)) { inputOptions.setSourcePaths(Arrays.asList(getSourceSnapshotPath(sourceDir, inputOptions.getToSnapshot()))); return false; } Path tmpDir = null; try { tmpDir = createTargetTmpDir(targetFs, targetDir); DiffInfo[] diffs = getDiffs(inputOptions, sourceFs, sourceDir, targetDir); if (diffs == null) { return false; } syncDiff(diffs, 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()))); } }    }