@Test public void testGetCopyableFiles() throws Exception { String sourceDir = getClass().getClassLoader().getResource("configBasedDatasetTest/src").getFile(); String destinationDir = getClass().getClassLoader().getResource("configBasedDatasetTest/dest").getFile(); long sourceWatermark = 100L; Collection<? extends CopyEntity> copyableFiles = testGetCopyableFilesHelper(sourceDir, destinationDir, sourceWatermark, false); Assert.assertEquals(copyableFiles.size(), 8); copyableFiles = testGetCopyableFilesHelper(sourceDir, destinationDir, sourceWatermark, true); Assert.assertEquals(copyableFiles.size(), 6); Set<Path> paths = Sets.newHashSet(new Path("dir1/file2"), new Path("dir1/file1"), new Path("dir2/file1"), new Path("dir2/file3")); for (CopyEntity copyEntity : copyableFiles) { if (copyEntity instanceof CopyableFile) { CopyableFile file = (CopyableFile) copyEntity; Path originRelativePath = PathUtils.relativizePath(PathUtils.getPathWithoutSchemeAndAuthority(file.getOrigin().getPath()), PathUtils.getPathWithoutSchemeAndAuthority(new Path(sourceDir))); Path targetRelativePath = PathUtils.relativizePath(PathUtils.getPathWithoutSchemeAndAuthority(file.getDestination()), PathUtils.getPathWithoutSchemeAndAuthority(new Path(destinationDir))); Assert.assertTrue(paths.contains(originRelativePath)); Assert.assertTrue(paths.contains(targetRelativePath)); Assert.assertEquals(originRelativePath, targetRelativePath); } else if (copyEntity instanceof PrePublishStep) { PrePublishStep pre = (PrePublishStep) copyEntity; Assert.assertTrue(pre.getStep() instanceof DeleteFileCommitStep); Assert.assertTrue(pre.explain().indexOf("configBasedDatasetTest/dest/dir1/file1") > 0); } else if (copyEntity instanceof PostPublishStep) { PostPublishStep post = (PostPublishStep) copyEntity; Assert.assertTrue(post.getStep() instanceof WatermarkMetadataGenerationCommitStep); Assert.assertTrue( post.explain().indexOf("dest/_metadata") > 0 && post.explain().indexOf("" + sourceWatermark) > 0); } else { throw new Exception("Wrong type"); } } }
@Override public Collection<? extends CopyEntity> getCopyableFiles(FileSystem targetFs, CopyConfiguration copyConfiguration) throws IOException { boolean enforceFileSizeMatch = this.rc.getEnforceFileSizeMatchFromConfigStore().isPresent()? this.rc.getEnforceFileSizeMatchFromConfigStore().get() : copyConfiguration.isEnforceFileLengthMatch(); List<CopyEntity> copyableFiles = Lists.newArrayList(); EndPoint copyFromRaw = copyRoute.getCopyFrom(); EndPoint copyToRaw = copyRoute.getCopyTo(); if (!(copyFromRaw instanceof HadoopFsEndPoint && copyToRaw instanceof HadoopFsEndPoint)) { log.warn("Currently only handle the Hadoop Fs EndPoint replication"); return copyableFiles; } if (!this.srcDataFileVersionStrategy.isPresent() || !this.dstDataFileVersionStrategy.isPresent()) { log.warn("Version strategy doesn't exist, cannot handle copy"); return copyableFiles; } if (!this.srcDataFileVersionStrategy.get().getClass().getName() .equals(this.dstDataFileVersionStrategy.get().getClass().getName())) { log.warn("Version strategy src: {} and dst: {} doesn't match, cannot handle copy.", this.srcDataFileVersionStrategy.get().getClass().getName(), this.dstDataFileVersionStrategy.get().getClass().getName()); return copyableFiles; } HadoopFsEndPoint copyFrom = (HadoopFsEndPoint) copyFromRaw; HadoopFsEndPoint copyTo = (HadoopFsEndPoint) copyToRaw; copyFrom.setPathFilter(pathFilter); copyFrom.setApplyFilterToDirectories(applyFilterToDirectories); copyTo.setPathFilter(pathFilter); copyTo.setApplyFilterToDirectories(applyFilterToDirectories); if (this.watermarkEnabled) { if ((!copyFromRaw.getWatermark().isPresent() && copyToRaw.getWatermark().isPresent()) || ( copyFromRaw.getWatermark().isPresent() && copyToRaw.getWatermark().isPresent() && copyFromRaw.getWatermark().get().compareTo(copyToRaw.getWatermark().get()) <= 0)) { log.info( "No need to copy as destination watermark >= source watermark with source watermark {}, for dataset with metadata {}", copyFromRaw.getWatermark().isPresent() ? copyFromRaw.getWatermark().get().toJson() : "N/A", this.rc.getMetaData()); return copyableFiles; } } Configuration conf = HadoopUtils.newConfiguration(); FileSystem copyFromFs = FileSystem.get(copyFrom.getFsURI(), conf); FileSystem copyToFs = FileSystem.get(copyTo.getFsURI(), conf); Collection<FileStatus> allFilesInSource = copyFrom.getFiles(); Collection<FileStatus> allFilesInTarget = copyTo.getFiles(); Set<FileStatus> copyFromFileStatuses = Sets.newHashSet(allFilesInSource); Map<Path, FileStatus> copyToFileMap = Maps.newHashMap(); for (FileStatus f : allFilesInTarget) { copyToFileMap.put(PathUtils.getPathWithoutSchemeAndAuthority(f.getPath()), f); } Collection<Path> deletedPaths = Lists.newArrayList(); boolean watermarkMetadataCopied = false; boolean deleteTargetIfNotExistOnSource = rc.isDeleteTargetIfNotExistOnSource(); for (FileStatus originFileStatus : copyFromFileStatuses) { Path relative = PathUtils.relativizePath(PathUtils.getPathWithoutSchemeAndAuthority(originFileStatus.getPath()), PathUtils.getPathWithoutSchemeAndAuthority(copyFrom.getDatasetPath())); Path newPath = new Path(copyTo.getDatasetPath(), relative); if (relative.toString().equals(ReplicaHadoopFsEndPoint.WATERMARK_FILE)) { watermarkMetadataCopied = true; } boolean shouldCopy = true; boolean useDirectGetModTime = this.srcDataFileVersionStrategy.isPresent() && this.srcDataFileVersionStrategy.get().getClass().getName().equals( ModTimeDataFileVersionStrategy.class.getName()); if (copyToFileMap.containsKey(newPath)) { Comparable srcVer = useDirectGetModTime ? originFileStatus.getModificationTime() : this.srcDataFileVersionStrategy.get().getVersion(originFileStatus.getPath()); Comparable dstVer = useDirectGetModTime ? copyToFileMap.get(newPath).getModificationTime() : this.dstDataFileVersionStrategy.get().getVersion(copyToFileMap.get(newPath).getPath()); if (srcVer.compareTo(dstVer) <= 0) { if (!enforceFileSizeMatch || copyToFileMap.get(newPath).getLen() == originFileStatus.getLen()) { log.debug("Copy from src {} (v:{}) to dst {} (v:{}) can be skipped.", originFileStatus.getPath(), srcVer, copyToFileMap.get(newPath).getPath(), dstVer); shouldCopy = false; } else { log.debug("Copy from src {} (v:{}) to dst {} (v:{}) can not be skipped due to unmatched file length.", originFileStatus.getPath(), srcVer, copyToFileMap.get(newPath).getPath(), dstVer); } } else { log.debug("Copy from src {} (v:{}) to dst {} (v:{}) is needed due to a higher version.", originFileStatus.getPath(), srcVer, copyToFileMap.get(newPath).getPath(), dstVer); } } else { log.debug("Copy from src {} to dst {} is needed because dst doesn't contain the file", originFileStatus.getPath(), copyToFileMap.get(newPath)); } if (shouldCopy) { if (copyToFileMap.containsKey(newPath)) { deletedPaths.add(newPath); } CopyableFile copyableFile = CopyableFile .fromOriginAndDestination(copyFromFs, originFileStatus, copyToFs.makeQualified(newPath), copyConfiguration) .fileSet(PathUtils.getPathWithoutSchemeAndAuthority(copyTo.getDatasetPath()).toString()) .dataFileVersionStrategy(this.versionStrategyFromCS.isPresent()? this.versionStrategyFromCS.get(): null) .build(); copyableFile.setFsDatasets(copyFromFs, copyToFs); copyableFiles.add(copyableFile); } copyToFileMap.remove(newPath); } if (deleteTargetIfNotExistOnSource) { deletedPaths.addAll(copyToFileMap.keySet()); } if (!deletedPaths.isEmpty()) { DeleteFileCommitStep deleteCommitStep = DeleteFileCommitStep.fromPaths(copyToFs, deletedPaths, this.props); copyableFiles.add( new PrePublishStep(copyTo.getDatasetPath().toString(), Maps.<String, String>newHashMap(), deleteCommitStep, 0)); } if ((!watermarkMetadataCopied) && copyFrom.getWatermark().isPresent()) { copyableFiles.add(new PostPublishStep(copyTo.getDatasetPath().toString(), Maps.<String, String>newHashMap(), new WatermarkMetadataGenerationCommitStep(copyTo.getFsURI().toString(), copyTo.getDatasetPath(), copyFrom.getWatermark().get()), 1)); } return copyableFiles; }
ConfigBasedDataset implements CopyableDataset { @Override public Collection<? extends CopyEntity> getCopyableFiles(FileSystem targetFs, CopyConfiguration copyConfiguration) throws IOException { boolean enforceFileSizeMatch = this.rc.getEnforceFileSizeMatchFromConfigStore().isPresent()? this.rc.getEnforceFileSizeMatchFromConfigStore().get() : copyConfiguration.isEnforceFileLengthMatch(); List<CopyEntity> copyableFiles = Lists.newArrayList(); EndPoint copyFromRaw = copyRoute.getCopyFrom(); EndPoint copyToRaw = copyRoute.getCopyTo(); if (!(copyFromRaw instanceof HadoopFsEndPoint && copyToRaw instanceof HadoopFsEndPoint)) { log.warn("Currently only handle the Hadoop Fs EndPoint replication"); return copyableFiles; } if (!this.srcDataFileVersionStrategy.isPresent() || !this.dstDataFileVersionStrategy.isPresent()) { log.warn("Version strategy doesn't exist, cannot handle copy"); return copyableFiles; } if (!this.srcDataFileVersionStrategy.get().getClass().getName() .equals(this.dstDataFileVersionStrategy.get().getClass().getName())) { log.warn("Version strategy src: {} and dst: {} doesn't match, cannot handle copy.", this.srcDataFileVersionStrategy.get().getClass().getName(), this.dstDataFileVersionStrategy.get().getClass().getName()); return copyableFiles; } HadoopFsEndPoint copyFrom = (HadoopFsEndPoint) copyFromRaw; HadoopFsEndPoint copyTo = (HadoopFsEndPoint) copyToRaw; copyFrom.setPathFilter(pathFilter); copyFrom.setApplyFilterToDirectories(applyFilterToDirectories); copyTo.setPathFilter(pathFilter); copyTo.setApplyFilterToDirectories(applyFilterToDirectories); if (this.watermarkEnabled) { if ((!copyFromRaw.getWatermark().isPresent() && copyToRaw.getWatermark().isPresent()) || ( copyFromRaw.getWatermark().isPresent() && copyToRaw.getWatermark().isPresent() && copyFromRaw.getWatermark().get().compareTo(copyToRaw.getWatermark().get()) <= 0)) { log.info( "No need to copy as destination watermark >= source watermark with source watermark {}, for dataset with metadata {}", copyFromRaw.getWatermark().isPresent() ? copyFromRaw.getWatermark().get().toJson() : "N/A", this.rc.getMetaData()); return copyableFiles; } } Configuration conf = HadoopUtils.newConfiguration(); FileSystem copyFromFs = FileSystem.get(copyFrom.getFsURI(), conf); FileSystem copyToFs = FileSystem.get(copyTo.getFsURI(), conf); Collection<FileStatus> allFilesInSource = copyFrom.getFiles(); Collection<FileStatus> allFilesInTarget = copyTo.getFiles(); Set<FileStatus> copyFromFileStatuses = Sets.newHashSet(allFilesInSource); Map<Path, FileStatus> copyToFileMap = Maps.newHashMap(); for (FileStatus f : allFilesInTarget) { copyToFileMap.put(PathUtils.getPathWithoutSchemeAndAuthority(f.getPath()), f); } Collection<Path> deletedPaths = Lists.newArrayList(); boolean watermarkMetadataCopied = false; boolean deleteTargetIfNotExistOnSource = rc.isDeleteTargetIfNotExistOnSource(); for (FileStatus originFileStatus : copyFromFileStatuses) { Path relative = PathUtils.relativizePath(PathUtils.getPathWithoutSchemeAndAuthority(originFileStatus.getPath()), PathUtils.getPathWithoutSchemeAndAuthority(copyFrom.getDatasetPath())); Path newPath = new Path(copyTo.getDatasetPath(), relative); if (relative.toString().equals(ReplicaHadoopFsEndPoint.WATERMARK_FILE)) { watermarkMetadataCopied = true; } boolean shouldCopy = true; boolean useDirectGetModTime = this.srcDataFileVersionStrategy.isPresent() && this.srcDataFileVersionStrategy.get().getClass().getName().equals( ModTimeDataFileVersionStrategy.class.getName()); if (copyToFileMap.containsKey(newPath)) { Comparable srcVer = useDirectGetModTime ? originFileStatus.getModificationTime() : this.srcDataFileVersionStrategy.get().getVersion(originFileStatus.getPath()); Comparable dstVer = useDirectGetModTime ? copyToFileMap.get(newPath).getModificationTime() : this.dstDataFileVersionStrategy.get().getVersion(copyToFileMap.get(newPath).getPath()); if (srcVer.compareTo(dstVer) <= 0) { if (!enforceFileSizeMatch || copyToFileMap.get(newPath).getLen() == originFileStatus.getLen()) { log.debug("Copy from src {} (v:{}) to dst {} (v:{}) can be skipped.", originFileStatus.getPath(), srcVer, copyToFileMap.get(newPath).getPath(), dstVer); shouldCopy = false; } else { log.debug("Copy from src {} (v:{}) to dst {} (v:{}) can not be skipped due to unmatched file length.", originFileStatus.getPath(), srcVer, copyToFileMap.get(newPath).getPath(), dstVer); } } else { log.debug("Copy from src {} (v:{}) to dst {} (v:{}) is needed due to a higher version.", originFileStatus.getPath(), srcVer, copyToFileMap.get(newPath).getPath(), dstVer); } } else { log.debug("Copy from src {} to dst {} is needed because dst doesn't contain the file", originFileStatus.getPath(), copyToFileMap.get(newPath)); } if (shouldCopy) { if (copyToFileMap.containsKey(newPath)) { deletedPaths.add(newPath); } CopyableFile copyableFile = CopyableFile .fromOriginAndDestination(copyFromFs, originFileStatus, copyToFs.makeQualified(newPath), copyConfiguration) .fileSet(PathUtils.getPathWithoutSchemeAndAuthority(copyTo.getDatasetPath()).toString()) .dataFileVersionStrategy(this.versionStrategyFromCS.isPresent()? this.versionStrategyFromCS.get(): null) .build(); copyableFile.setFsDatasets(copyFromFs, copyToFs); copyableFiles.add(copyableFile); } copyToFileMap.remove(newPath); } if (deleteTargetIfNotExistOnSource) { deletedPaths.addAll(copyToFileMap.keySet()); } if (!deletedPaths.isEmpty()) { DeleteFileCommitStep deleteCommitStep = DeleteFileCommitStep.fromPaths(copyToFs, deletedPaths, this.props); copyableFiles.add( new PrePublishStep(copyTo.getDatasetPath().toString(), Maps.<String, String>newHashMap(), deleteCommitStep, 0)); } if ((!watermarkMetadataCopied) && copyFrom.getWatermark().isPresent()) { copyableFiles.add(new PostPublishStep(copyTo.getDatasetPath().toString(), Maps.<String, String>newHashMap(), new WatermarkMetadataGenerationCommitStep(copyTo.getFsURI().toString(), copyTo.getDatasetPath(), copyFrom.getWatermark().get()), 1)); } return copyableFiles; } }
ConfigBasedDataset implements CopyableDataset { @Override public Collection<? extends CopyEntity> getCopyableFiles(FileSystem targetFs, CopyConfiguration copyConfiguration) throws IOException { boolean enforceFileSizeMatch = this.rc.getEnforceFileSizeMatchFromConfigStore().isPresent()? this.rc.getEnforceFileSizeMatchFromConfigStore().get() : copyConfiguration.isEnforceFileLengthMatch(); List<CopyEntity> copyableFiles = Lists.newArrayList(); EndPoint copyFromRaw = copyRoute.getCopyFrom(); EndPoint copyToRaw = copyRoute.getCopyTo(); if (!(copyFromRaw instanceof HadoopFsEndPoint && copyToRaw instanceof HadoopFsEndPoint)) { log.warn("Currently only handle the Hadoop Fs EndPoint replication"); return copyableFiles; } if (!this.srcDataFileVersionStrategy.isPresent() || !this.dstDataFileVersionStrategy.isPresent()) { log.warn("Version strategy doesn't exist, cannot handle copy"); return copyableFiles; } if (!this.srcDataFileVersionStrategy.get().getClass().getName() .equals(this.dstDataFileVersionStrategy.get().getClass().getName())) { log.warn("Version strategy src: {} and dst: {} doesn't match, cannot handle copy.", this.srcDataFileVersionStrategy.get().getClass().getName(), this.dstDataFileVersionStrategy.get().getClass().getName()); return copyableFiles; } HadoopFsEndPoint copyFrom = (HadoopFsEndPoint) copyFromRaw; HadoopFsEndPoint copyTo = (HadoopFsEndPoint) copyToRaw; copyFrom.setPathFilter(pathFilter); copyFrom.setApplyFilterToDirectories(applyFilterToDirectories); copyTo.setPathFilter(pathFilter); copyTo.setApplyFilterToDirectories(applyFilterToDirectories); if (this.watermarkEnabled) { if ((!copyFromRaw.getWatermark().isPresent() && copyToRaw.getWatermark().isPresent()) || ( copyFromRaw.getWatermark().isPresent() && copyToRaw.getWatermark().isPresent() && copyFromRaw.getWatermark().get().compareTo(copyToRaw.getWatermark().get()) <= 0)) { log.info( "No need to copy as destination watermark >= source watermark with source watermark {}, for dataset with metadata {}", copyFromRaw.getWatermark().isPresent() ? copyFromRaw.getWatermark().get().toJson() : "N/A", this.rc.getMetaData()); return copyableFiles; } } Configuration conf = HadoopUtils.newConfiguration(); FileSystem copyFromFs = FileSystem.get(copyFrom.getFsURI(), conf); FileSystem copyToFs = FileSystem.get(copyTo.getFsURI(), conf); Collection<FileStatus> allFilesInSource = copyFrom.getFiles(); Collection<FileStatus> allFilesInTarget = copyTo.getFiles(); Set<FileStatus> copyFromFileStatuses = Sets.newHashSet(allFilesInSource); Map<Path, FileStatus> copyToFileMap = Maps.newHashMap(); for (FileStatus f : allFilesInTarget) { copyToFileMap.put(PathUtils.getPathWithoutSchemeAndAuthority(f.getPath()), f); } Collection<Path> deletedPaths = Lists.newArrayList(); boolean watermarkMetadataCopied = false; boolean deleteTargetIfNotExistOnSource = rc.isDeleteTargetIfNotExistOnSource(); for (FileStatus originFileStatus : copyFromFileStatuses) { Path relative = PathUtils.relativizePath(PathUtils.getPathWithoutSchemeAndAuthority(originFileStatus.getPath()), PathUtils.getPathWithoutSchemeAndAuthority(copyFrom.getDatasetPath())); Path newPath = new Path(copyTo.getDatasetPath(), relative); if (relative.toString().equals(ReplicaHadoopFsEndPoint.WATERMARK_FILE)) { watermarkMetadataCopied = true; } boolean shouldCopy = true; boolean useDirectGetModTime = this.srcDataFileVersionStrategy.isPresent() && this.srcDataFileVersionStrategy.get().getClass().getName().equals( ModTimeDataFileVersionStrategy.class.getName()); if (copyToFileMap.containsKey(newPath)) { Comparable srcVer = useDirectGetModTime ? originFileStatus.getModificationTime() : this.srcDataFileVersionStrategy.get().getVersion(originFileStatus.getPath()); Comparable dstVer = useDirectGetModTime ? copyToFileMap.get(newPath).getModificationTime() : this.dstDataFileVersionStrategy.get().getVersion(copyToFileMap.get(newPath).getPath()); if (srcVer.compareTo(dstVer) <= 0) { if (!enforceFileSizeMatch || copyToFileMap.get(newPath).getLen() == originFileStatus.getLen()) { log.debug("Copy from src {} (v:{}) to dst {} (v:{}) can be skipped.", originFileStatus.getPath(), srcVer, copyToFileMap.get(newPath).getPath(), dstVer); shouldCopy = false; } else { log.debug("Copy from src {} (v:{}) to dst {} (v:{}) can not be skipped due to unmatched file length.", originFileStatus.getPath(), srcVer, copyToFileMap.get(newPath).getPath(), dstVer); } } else { log.debug("Copy from src {} (v:{}) to dst {} (v:{}) is needed due to a higher version.", originFileStatus.getPath(), srcVer, copyToFileMap.get(newPath).getPath(), dstVer); } } else { log.debug("Copy from src {} to dst {} is needed because dst doesn't contain the file", originFileStatus.getPath(), copyToFileMap.get(newPath)); } if (shouldCopy) { if (copyToFileMap.containsKey(newPath)) { deletedPaths.add(newPath); } CopyableFile copyableFile = CopyableFile .fromOriginAndDestination(copyFromFs, originFileStatus, copyToFs.makeQualified(newPath), copyConfiguration) .fileSet(PathUtils.getPathWithoutSchemeAndAuthority(copyTo.getDatasetPath()).toString()) .dataFileVersionStrategy(this.versionStrategyFromCS.isPresent()? this.versionStrategyFromCS.get(): null) .build(); copyableFile.setFsDatasets(copyFromFs, copyToFs); copyableFiles.add(copyableFile); } copyToFileMap.remove(newPath); } if (deleteTargetIfNotExistOnSource) { deletedPaths.addAll(copyToFileMap.keySet()); } if (!deletedPaths.isEmpty()) { DeleteFileCommitStep deleteCommitStep = DeleteFileCommitStep.fromPaths(copyToFs, deletedPaths, this.props); copyableFiles.add( new PrePublishStep(copyTo.getDatasetPath().toString(), Maps.<String, String>newHashMap(), deleteCommitStep, 0)); } if ((!watermarkMetadataCopied) && copyFrom.getWatermark().isPresent()) { copyableFiles.add(new PostPublishStep(copyTo.getDatasetPath().toString(), Maps.<String, String>newHashMap(), new WatermarkMetadataGenerationCommitStep(copyTo.getFsURI().toString(), copyTo.getDatasetPath(), copyFrom.getWatermark().get()), 1)); } return copyableFiles; } ConfigBasedDataset(ReplicationConfiguration rc, Properties props, CopyRoute copyRoute);  ConfigBasedDataset(ReplicationConfiguration rc, Properties props, CopyRoute copyRoute, String datasetURN); }
ConfigBasedDataset implements CopyableDataset { @Override public Collection<? extends CopyEntity> getCopyableFiles(FileSystem targetFs, CopyConfiguration copyConfiguration) throws IOException { boolean enforceFileSizeMatch = this.rc.getEnforceFileSizeMatchFromConfigStore().isPresent()? this.rc.getEnforceFileSizeMatchFromConfigStore().get() : copyConfiguration.isEnforceFileLengthMatch(); List<CopyEntity> copyableFiles = Lists.newArrayList(); EndPoint copyFromRaw = copyRoute.getCopyFrom(); EndPoint copyToRaw = copyRoute.getCopyTo(); if (!(copyFromRaw instanceof HadoopFsEndPoint && copyToRaw instanceof HadoopFsEndPoint)) { log.warn("Currently only handle the Hadoop Fs EndPoint replication"); return copyableFiles; } if (!this.srcDataFileVersionStrategy.isPresent() || !this.dstDataFileVersionStrategy.isPresent()) { log.warn("Version strategy doesn't exist, cannot handle copy"); return copyableFiles; } if (!this.srcDataFileVersionStrategy.get().getClass().getName() .equals(this.dstDataFileVersionStrategy.get().getClass().getName())) { log.warn("Version strategy src: {} and dst: {} doesn't match, cannot handle copy.", this.srcDataFileVersionStrategy.get().getClass().getName(), this.dstDataFileVersionStrategy.get().getClass().getName()); return copyableFiles; } HadoopFsEndPoint copyFrom = (HadoopFsEndPoint) copyFromRaw; HadoopFsEndPoint copyTo = (HadoopFsEndPoint) copyToRaw; copyFrom.setPathFilter(pathFilter); copyFrom.setApplyFilterToDirectories(applyFilterToDirectories); copyTo.setPathFilter(pathFilter); copyTo.setApplyFilterToDirectories(applyFilterToDirectories); if (this.watermarkEnabled) { if ((!copyFromRaw.getWatermark().isPresent() && copyToRaw.getWatermark().isPresent()) || ( copyFromRaw.getWatermark().isPresent() && copyToRaw.getWatermark().isPresent() && copyFromRaw.getWatermark().get().compareTo(copyToRaw.getWatermark().get()) <= 0)) { log.info( "No need to copy as destination watermark >= source watermark with source watermark {}, for dataset with metadata {}", copyFromRaw.getWatermark().isPresent() ? copyFromRaw.getWatermark().get().toJson() : "N/A", this.rc.getMetaData()); return copyableFiles; } } Configuration conf = HadoopUtils.newConfiguration(); FileSystem copyFromFs = FileSystem.get(copyFrom.getFsURI(), conf); FileSystem copyToFs = FileSystem.get(copyTo.getFsURI(), conf); Collection<FileStatus> allFilesInSource = copyFrom.getFiles(); Collection<FileStatus> allFilesInTarget = copyTo.getFiles(); Set<FileStatus> copyFromFileStatuses = Sets.newHashSet(allFilesInSource); Map<Path, FileStatus> copyToFileMap = Maps.newHashMap(); for (FileStatus f : allFilesInTarget) { copyToFileMap.put(PathUtils.getPathWithoutSchemeAndAuthority(f.getPath()), f); } Collection<Path> deletedPaths = Lists.newArrayList(); boolean watermarkMetadataCopied = false; boolean deleteTargetIfNotExistOnSource = rc.isDeleteTargetIfNotExistOnSource(); for (FileStatus originFileStatus : copyFromFileStatuses) { Path relative = PathUtils.relativizePath(PathUtils.getPathWithoutSchemeAndAuthority(originFileStatus.getPath()), PathUtils.getPathWithoutSchemeAndAuthority(copyFrom.getDatasetPath())); Path newPath = new Path(copyTo.getDatasetPath(), relative); if (relative.toString().equals(ReplicaHadoopFsEndPoint.WATERMARK_FILE)) { watermarkMetadataCopied = true; } boolean shouldCopy = true; boolean useDirectGetModTime = this.srcDataFileVersionStrategy.isPresent() && this.srcDataFileVersionStrategy.get().getClass().getName().equals( ModTimeDataFileVersionStrategy.class.getName()); if (copyToFileMap.containsKey(newPath)) { Comparable srcVer = useDirectGetModTime ? originFileStatus.getModificationTime() : this.srcDataFileVersionStrategy.get().getVersion(originFileStatus.getPath()); Comparable dstVer = useDirectGetModTime ? copyToFileMap.get(newPath).getModificationTime() : this.dstDataFileVersionStrategy.get().getVersion(copyToFileMap.get(newPath).getPath()); if (srcVer.compareTo(dstVer) <= 0) { if (!enforceFileSizeMatch || copyToFileMap.get(newPath).getLen() == originFileStatus.getLen()) { log.debug("Copy from src {} (v:{}) to dst {} (v:{}) can be skipped.", originFileStatus.getPath(), srcVer, copyToFileMap.get(newPath).getPath(), dstVer); shouldCopy = false; } else { log.debug("Copy from src {} (v:{}) to dst {} (v:{}) can not be skipped due to unmatched file length.", originFileStatus.getPath(), srcVer, copyToFileMap.get(newPath).getPath(), dstVer); } } else { log.debug("Copy from src {} (v:{}) to dst {} (v:{}) is needed due to a higher version.", originFileStatus.getPath(), srcVer, copyToFileMap.get(newPath).getPath(), dstVer); } } else { log.debug("Copy from src {} to dst {} is needed because dst doesn't contain the file", originFileStatus.getPath(), copyToFileMap.get(newPath)); } if (shouldCopy) { if (copyToFileMap.containsKey(newPath)) { deletedPaths.add(newPath); } CopyableFile copyableFile = CopyableFile .fromOriginAndDestination(copyFromFs, originFileStatus, copyToFs.makeQualified(newPath), copyConfiguration) .fileSet(PathUtils.getPathWithoutSchemeAndAuthority(copyTo.getDatasetPath()).toString()) .dataFileVersionStrategy(this.versionStrategyFromCS.isPresent()? this.versionStrategyFromCS.get(): null) .build(); copyableFile.setFsDatasets(copyFromFs, copyToFs); copyableFiles.add(copyableFile); } copyToFileMap.remove(newPath); } if (deleteTargetIfNotExistOnSource) { deletedPaths.addAll(copyToFileMap.keySet()); } if (!deletedPaths.isEmpty()) { DeleteFileCommitStep deleteCommitStep = DeleteFileCommitStep.fromPaths(copyToFs, deletedPaths, this.props); copyableFiles.add( new PrePublishStep(copyTo.getDatasetPath().toString(), Maps.<String, String>newHashMap(), deleteCommitStep, 0)); } if ((!watermarkMetadataCopied) && copyFrom.getWatermark().isPresent()) { copyableFiles.add(new PostPublishStep(copyTo.getDatasetPath().toString(), Maps.<String, String>newHashMap(), new WatermarkMetadataGenerationCommitStep(copyTo.getFsURI().toString(), copyTo.getDatasetPath(), copyFrom.getWatermark().get()), 1)); } return copyableFiles; } ConfigBasedDataset(ReplicationConfiguration rc, Properties props, CopyRoute copyRoute);  ConfigBasedDataset(ReplicationConfiguration rc, Properties props, CopyRoute copyRoute, String datasetURN); boolean schemaCheckEnabled(); @Override String datasetURN(); @Override Collection<? extends CopyEntity> getCopyableFiles(FileSystem targetFs, CopyConfiguration copyConfiguration); }
ConfigBasedDataset implements CopyableDataset { @Override public Collection<? extends CopyEntity> getCopyableFiles(FileSystem targetFs, CopyConfiguration copyConfiguration) throws IOException { boolean enforceFileSizeMatch = this.rc.getEnforceFileSizeMatchFromConfigStore().isPresent()? this.rc.getEnforceFileSizeMatchFromConfigStore().get() : copyConfiguration.isEnforceFileLengthMatch(); List<CopyEntity> copyableFiles = Lists.newArrayList(); EndPoint copyFromRaw = copyRoute.getCopyFrom(); EndPoint copyToRaw = copyRoute.getCopyTo(); if (!(copyFromRaw instanceof HadoopFsEndPoint && copyToRaw instanceof HadoopFsEndPoint)) { log.warn("Currently only handle the Hadoop Fs EndPoint replication"); return copyableFiles; } if (!this.srcDataFileVersionStrategy.isPresent() || !this.dstDataFileVersionStrategy.isPresent()) { log.warn("Version strategy doesn't exist, cannot handle copy"); return copyableFiles; } if (!this.srcDataFileVersionStrategy.get().getClass().getName() .equals(this.dstDataFileVersionStrategy.get().getClass().getName())) { log.warn("Version strategy src: {} and dst: {} doesn't match, cannot handle copy.", this.srcDataFileVersionStrategy.get().getClass().getName(), this.dstDataFileVersionStrategy.get().getClass().getName()); return copyableFiles; } HadoopFsEndPoint copyFrom = (HadoopFsEndPoint) copyFromRaw; HadoopFsEndPoint copyTo = (HadoopFsEndPoint) copyToRaw; copyFrom.setPathFilter(pathFilter); copyFrom.setApplyFilterToDirectories(applyFilterToDirectories); copyTo.setPathFilter(pathFilter); copyTo.setApplyFilterToDirectories(applyFilterToDirectories); if (this.watermarkEnabled) { if ((!copyFromRaw.getWatermark().isPresent() && copyToRaw.getWatermark().isPresent()) || ( copyFromRaw.getWatermark().isPresent() && copyToRaw.getWatermark().isPresent() && copyFromRaw.getWatermark().get().compareTo(copyToRaw.getWatermark().get()) <= 0)) { log.info( "No need to copy as destination watermark >= source watermark with source watermark {}, for dataset with metadata {}", copyFromRaw.getWatermark().isPresent() ? copyFromRaw.getWatermark().get().toJson() : "N/A", this.rc.getMetaData()); return copyableFiles; } } Configuration conf = HadoopUtils.newConfiguration(); FileSystem copyFromFs = FileSystem.get(copyFrom.getFsURI(), conf); FileSystem copyToFs = FileSystem.get(copyTo.getFsURI(), conf); Collection<FileStatus> allFilesInSource = copyFrom.getFiles(); Collection<FileStatus> allFilesInTarget = copyTo.getFiles(); Set<FileStatus> copyFromFileStatuses = Sets.newHashSet(allFilesInSource); Map<Path, FileStatus> copyToFileMap = Maps.newHashMap(); for (FileStatus f : allFilesInTarget) { copyToFileMap.put(PathUtils.getPathWithoutSchemeAndAuthority(f.getPath()), f); } Collection<Path> deletedPaths = Lists.newArrayList(); boolean watermarkMetadataCopied = false; boolean deleteTargetIfNotExistOnSource = rc.isDeleteTargetIfNotExistOnSource(); for (FileStatus originFileStatus : copyFromFileStatuses) { Path relative = PathUtils.relativizePath(PathUtils.getPathWithoutSchemeAndAuthority(originFileStatus.getPath()), PathUtils.getPathWithoutSchemeAndAuthority(copyFrom.getDatasetPath())); Path newPath = new Path(copyTo.getDatasetPath(), relative); if (relative.toString().equals(ReplicaHadoopFsEndPoint.WATERMARK_FILE)) { watermarkMetadataCopied = true; } boolean shouldCopy = true; boolean useDirectGetModTime = this.srcDataFileVersionStrategy.isPresent() && this.srcDataFileVersionStrategy.get().getClass().getName().equals( ModTimeDataFileVersionStrategy.class.getName()); if (copyToFileMap.containsKey(newPath)) { Comparable srcVer = useDirectGetModTime ? originFileStatus.getModificationTime() : this.srcDataFileVersionStrategy.get().getVersion(originFileStatus.getPath()); Comparable dstVer = useDirectGetModTime ? copyToFileMap.get(newPath).getModificationTime() : this.dstDataFileVersionStrategy.get().getVersion(copyToFileMap.get(newPath).getPath()); if (srcVer.compareTo(dstVer) <= 0) { if (!enforceFileSizeMatch || copyToFileMap.get(newPath).getLen() == originFileStatus.getLen()) { log.debug("Copy from src {} (v:{}) to dst {} (v:{}) can be skipped.", originFileStatus.getPath(), srcVer, copyToFileMap.get(newPath).getPath(), dstVer); shouldCopy = false; } else { log.debug("Copy from src {} (v:{}) to dst {} (v:{}) can not be skipped due to unmatched file length.", originFileStatus.getPath(), srcVer, copyToFileMap.get(newPath).getPath(), dstVer); } } else { log.debug("Copy from src {} (v:{}) to dst {} (v:{}) is needed due to a higher version.", originFileStatus.getPath(), srcVer, copyToFileMap.get(newPath).getPath(), dstVer); } } else { log.debug("Copy from src {} to dst {} is needed because dst doesn't contain the file", originFileStatus.getPath(), copyToFileMap.get(newPath)); } if (shouldCopy) { if (copyToFileMap.containsKey(newPath)) { deletedPaths.add(newPath); } CopyableFile copyableFile = CopyableFile .fromOriginAndDestination(copyFromFs, originFileStatus, copyToFs.makeQualified(newPath), copyConfiguration) .fileSet(PathUtils.getPathWithoutSchemeAndAuthority(copyTo.getDatasetPath()).toString()) .dataFileVersionStrategy(this.versionStrategyFromCS.isPresent()? this.versionStrategyFromCS.get(): null) .build(); copyableFile.setFsDatasets(copyFromFs, copyToFs); copyableFiles.add(copyableFile); } copyToFileMap.remove(newPath); } if (deleteTargetIfNotExistOnSource) { deletedPaths.addAll(copyToFileMap.keySet()); } if (!deletedPaths.isEmpty()) { DeleteFileCommitStep deleteCommitStep = DeleteFileCommitStep.fromPaths(copyToFs, deletedPaths, this.props); copyableFiles.add( new PrePublishStep(copyTo.getDatasetPath().toString(), Maps.<String, String>newHashMap(), deleteCommitStep, 0)); } if ((!watermarkMetadataCopied) && copyFrom.getWatermark().isPresent()) { copyableFiles.add(new PostPublishStep(copyTo.getDatasetPath().toString(), Maps.<String, String>newHashMap(), new WatermarkMetadataGenerationCommitStep(copyTo.getFsURI().toString(), copyTo.getDatasetPath(), copyFrom.getWatermark().get()), 1)); } return copyableFiles; } ConfigBasedDataset(ReplicationConfiguration rc, Properties props, CopyRoute copyRoute);  ConfigBasedDataset(ReplicationConfiguration rc, Properties props, CopyRoute copyRoute, String datasetURN); boolean schemaCheckEnabled(); @Override String datasetURN(); @Override Collection<? extends CopyEntity> getCopyableFiles(FileSystem targetFs, CopyConfiguration copyConfiguration);  }