@Test public void testGetCopyableFiles() throws IOException { Properties props = new Properties(); props.put(TimestampBasedCopyableDataset.COPY_POLICY, TimeBasedCopyPolicyForTest.class.getName()); props.put(TimestampBasedCopyableDataset.DATASET_VERSION_FINDER, TimestampedDatasetVersionFinderForTest.class.getName()); TimestampBasedCopyableDataset copyabledataset = new TimestampBasedCopyableDatasetForTest(localFs, props, new Path("/data/tracking/PVE")); Collection<CopyableFile> copyableFiles = copyabledataset.getCopyableFiles(localFs, null); Assert.assertEquals(copyableFiles.size(), 6); Pattern pattern = Pattern.compile("dummy/[\\d]*/file[12]"); Set<String> resultFilesets = Sets.newHashSet(); for (CopyableFile copyableFile : copyableFiles) { String copyableFileset = copyableFile.getFileSet(); Assert.assertTrue(pattern.matcher(copyableFileset).matches()); resultFilesets.add(copyableFileset); } Assert.assertEquals(resultFilesets.size(), 6); }
@Override public Collection<CopyableFile> getCopyableFiles(FileSystem targetFs, CopyConfiguration configuration) throws IOException { log.info(String.format("Getting copyable files at root path: %s", this.datasetRoot)); List<TimestampedDatasetVersion> versions = Lists.newArrayList(this.datasetVersionFinder.findDatasetVersions(this)); if (versions.isEmpty()) { log.warn("No dataset version can be found. Ignoring."); return Lists.newArrayList(); } Collection<TimestampedDatasetVersion> copyableVersions = this.versionSelectionPolicy.listSelectedVersions(versions); ConcurrentLinkedQueue<CopyableFile> copyableFileList = new ConcurrentLinkedQueue<>(); List<Future<?>> futures = Lists.newArrayList(); for (TimestampedDatasetVersion copyableVersion : copyableVersions) { futures.add(this.executor.submit(this.getCopyableFileGenetator(targetFs, configuration, copyableVersion, copyableFileList))); } try { for (Future<?> future : futures) { future.get(); } } catch (ExecutionException | InterruptedException e) { throw new IOException("Failed to generate copyable files.", e); } finally { ExecutorsUtils.shutdownExecutorService(executor, Optional.of(log)); } return copyableFileList; }
TimestampBasedCopyableDataset implements CopyableDataset, FileSystemDataset { @Override public Collection<CopyableFile> getCopyableFiles(FileSystem targetFs, CopyConfiguration configuration) throws IOException { log.info(String.format("Getting copyable files at root path: %s", this.datasetRoot)); List<TimestampedDatasetVersion> versions = Lists.newArrayList(this.datasetVersionFinder.findDatasetVersions(this)); if (versions.isEmpty()) { log.warn("No dataset version can be found. Ignoring."); return Lists.newArrayList(); } Collection<TimestampedDatasetVersion> copyableVersions = this.versionSelectionPolicy.listSelectedVersions(versions); ConcurrentLinkedQueue<CopyableFile> copyableFileList = new ConcurrentLinkedQueue<>(); List<Future<?>> futures = Lists.newArrayList(); for (TimestampedDatasetVersion copyableVersion : copyableVersions) { futures.add(this.executor.submit(this.getCopyableFileGenetator(targetFs, configuration, copyableVersion, copyableFileList))); } try { for (Future<?> future : futures) { future.get(); } } catch (ExecutionException | InterruptedException e) { throw new IOException("Failed to generate copyable files.", e); } finally { ExecutorsUtils.shutdownExecutorService(executor, Optional.of(log)); } return copyableFileList; } }
TimestampBasedCopyableDataset implements CopyableDataset, FileSystemDataset { @Override public Collection<CopyableFile> getCopyableFiles(FileSystem targetFs, CopyConfiguration configuration) throws IOException { log.info(String.format("Getting copyable files at root path: %s", this.datasetRoot)); List<TimestampedDatasetVersion> versions = Lists.newArrayList(this.datasetVersionFinder.findDatasetVersions(this)); if (versions.isEmpty()) { log.warn("No dataset version can be found. Ignoring."); return Lists.newArrayList(); } Collection<TimestampedDatasetVersion> copyableVersions = this.versionSelectionPolicy.listSelectedVersions(versions); ConcurrentLinkedQueue<CopyableFile> copyableFileList = new ConcurrentLinkedQueue<>(); List<Future<?>> futures = Lists.newArrayList(); for (TimestampedDatasetVersion copyableVersion : copyableVersions) { futures.add(this.executor.submit(this.getCopyableFileGenetator(targetFs, configuration, copyableVersion, copyableFileList))); } try { for (Future<?> future : futures) { future.get(); } } catch (ExecutionException | InterruptedException e) { throw new IOException("Failed to generate copyable files.", e); } finally { ExecutorsUtils.shutdownExecutorService(executor, Optional.of(log)); } return copyableFileList; } TimestampBasedCopyableDataset(FileSystem fs, Properties props, Path datasetRoot); }
TimestampBasedCopyableDataset implements CopyableDataset, FileSystemDataset { @Override public Collection<CopyableFile> getCopyableFiles(FileSystem targetFs, CopyConfiguration configuration) throws IOException { log.info(String.format("Getting copyable files at root path: %s", this.datasetRoot)); List<TimestampedDatasetVersion> versions = Lists.newArrayList(this.datasetVersionFinder.findDatasetVersions(this)); if (versions.isEmpty()) { log.warn("No dataset version can be found. Ignoring."); return Lists.newArrayList(); } Collection<TimestampedDatasetVersion> copyableVersions = this.versionSelectionPolicy.listSelectedVersions(versions); ConcurrentLinkedQueue<CopyableFile> copyableFileList = new ConcurrentLinkedQueue<>(); List<Future<?>> futures = Lists.newArrayList(); for (TimestampedDatasetVersion copyableVersion : copyableVersions) { futures.add(this.executor.submit(this.getCopyableFileGenetator(targetFs, configuration, copyableVersion, copyableFileList))); } try { for (Future<?> future : futures) { future.get(); } } catch (ExecutionException | InterruptedException e) { throw new IOException("Failed to generate copyable files.", e); } finally { ExecutorsUtils.shutdownExecutorService(executor, Optional.of(log)); } return copyableFileList; } TimestampBasedCopyableDataset(FileSystem fs, Properties props, Path datasetRoot); @Override Collection<CopyableFile> getCopyableFiles(FileSystem targetFs, CopyConfiguration configuration); @Override String datasetURN(); @Override Path datasetRoot(); }
TimestampBasedCopyableDataset implements CopyableDataset, FileSystemDataset { @Override public Collection<CopyableFile> getCopyableFiles(FileSystem targetFs, CopyConfiguration configuration) throws IOException { log.info(String.format("Getting copyable files at root path: %s", this.datasetRoot)); List<TimestampedDatasetVersion> versions = Lists.newArrayList(this.datasetVersionFinder.findDatasetVersions(this)); if (versions.isEmpty()) { log.warn("No dataset version can be found. Ignoring."); return Lists.newArrayList(); } Collection<TimestampedDatasetVersion> copyableVersions = this.versionSelectionPolicy.listSelectedVersions(versions); ConcurrentLinkedQueue<CopyableFile> copyableFileList = new ConcurrentLinkedQueue<>(); List<Future<?>> futures = Lists.newArrayList(); for (TimestampedDatasetVersion copyableVersion : copyableVersions) { futures.add(this.executor.submit(this.getCopyableFileGenetator(targetFs, configuration, copyableVersion, copyableFileList))); } try { for (Future<?> future : futures) { future.get(); } } catch (ExecutionException | InterruptedException e) { throw new IOException("Failed to generate copyable files.", e); } finally { ExecutorsUtils.shutdownExecutorService(executor, Optional.of(log)); } return copyableFileList; } TimestampBasedCopyableDataset(FileSystem fs, Properties props, Path datasetRoot); @Override Collection<CopyableFile> getCopyableFiles(FileSystem targetFs, CopyConfiguration configuration); @Override String datasetURN(); @Override Path datasetRoot(); static final String DATASET_VERSION_FINDER; static final String DEFAULT_DATASET_VERSION_FINDER; static final String COPY_POLICY; static final String DEFAULT_COPY_POLICY; static final String THREADPOOL_SIZE_TO_GET_COPYABLE_FILES; static final String DEFAULT_THREADPOOL_SIZE_TO_GET_COPYABLE_FILES; }