@Test @SuppressWarnings("unchecked") public void testPublishMultipleDatasets() throws Exception { State state = getTestState("testPublishMultipleDatasets"); state.setProp(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, "/"); Path testMethodTempPath = new Path(testClassTempPath, "testPublishMultipleDatasets"); CopyDataPublisher copyDataPublisher = closer.register(new CopyDataPublisher(state)); TestDatasetManager dataset1Manager = new TestDatasetManager(testMethodTempPath, state, "dataset1TargetPath", ImmutableList.of("a/b", "a/c", "d/e")); dataset1Manager.createDatasetFiles(); TestDatasetManager dataset2Manager = new TestDatasetManager(testMethodTempPath, state, "dataset2TargetPath", ImmutableList.of("a/b", "a/c", "d/e")); dataset2Manager.createDatasetFiles(); dataset1Manager.verifyDoesntExist(); dataset2Manager.verifyDoesntExist(); copyDataPublisher.publishData(combine(dataset1Manager.getWorkUnitStates(), dataset2Manager.getWorkUnitStates())); dataset1Manager.verifyExists(); dataset2Manager.verifyExists(); }
@Override public void publishData(Collection<? extends WorkUnitState> states) throws IOException { Multimap<CopyEntity.DatasetAndPartition, WorkUnitState> datasets = groupByFileSet(states); boolean allDatasetsPublished = true; for (CopyEntity.DatasetAndPartition datasetAndPartition : datasets.keySet()) { try { this.publishFileSet(datasetAndPartition, datasets.get(datasetAndPartition)); } catch (Throwable e) { CopyEventSubmitterHelper.submitFailedDatasetPublish(this.eventSubmitter, datasetAndPartition); log.error("Failed to publish " + datasetAndPartition.getDataset().getDatasetURN(), e); allDatasetsPublished = false; } } if (!allDatasetsPublished) { throw new IOException("Not all datasets published successfully"); } }
CopyDataPublisher extends DataPublisher implements UnpublishedHandling { @Override public void publishData(Collection<? extends WorkUnitState> states) throws IOException { Multimap<CopyEntity.DatasetAndPartition, WorkUnitState> datasets = groupByFileSet(states); boolean allDatasetsPublished = true; for (CopyEntity.DatasetAndPartition datasetAndPartition : datasets.keySet()) { try { this.publishFileSet(datasetAndPartition, datasets.get(datasetAndPartition)); } catch (Throwable e) { CopyEventSubmitterHelper.submitFailedDatasetPublish(this.eventSubmitter, datasetAndPartition); log.error("Failed to publish " + datasetAndPartition.getDataset().getDatasetURN(), e); allDatasetsPublished = false; } } if (!allDatasetsPublished) { throw new IOException("Not all datasets published successfully"); } } }
CopyDataPublisher extends DataPublisher implements UnpublishedHandling { @Override public void publishData(Collection<? extends WorkUnitState> states) throws IOException { Multimap<CopyEntity.DatasetAndPartition, WorkUnitState> datasets = groupByFileSet(states); boolean allDatasetsPublished = true; for (CopyEntity.DatasetAndPartition datasetAndPartition : datasets.keySet()) { try { this.publishFileSet(datasetAndPartition, datasets.get(datasetAndPartition)); } catch (Throwable e) { CopyEventSubmitterHelper.submitFailedDatasetPublish(this.eventSubmitter, datasetAndPartition); log.error("Failed to publish " + datasetAndPartition.getDataset().getDatasetURN(), e); allDatasetsPublished = false; } } if (!allDatasetsPublished) { throw new IOException("Not all datasets published successfully"); } } CopyDataPublisher(State state); }
CopyDataPublisher extends DataPublisher implements UnpublishedHandling { @Override public void publishData(Collection<? extends WorkUnitState> states) throws IOException { Multimap<CopyEntity.DatasetAndPartition, WorkUnitState> datasets = groupByFileSet(states); boolean allDatasetsPublished = true; for (CopyEntity.DatasetAndPartition datasetAndPartition : datasets.keySet()) { try { this.publishFileSet(datasetAndPartition, datasets.get(datasetAndPartition)); } catch (Throwable e) { CopyEventSubmitterHelper.submitFailedDatasetPublish(this.eventSubmitter, datasetAndPartition); log.error("Failed to publish " + datasetAndPartition.getDataset().getDatasetURN(), e); allDatasetsPublished = false; } } if (!allDatasetsPublished) { throw new IOException("Not all datasets published successfully"); } } CopyDataPublisher(State state); @Override boolean isThreadSafe(); @Override void publishData(Collection<? extends WorkUnitState> states); @Override void handleUnpublishedWorkUnits(Collection<? extends WorkUnitState> states); @Override void publishMetadata(Collection<? extends WorkUnitState> states); @Override void close(); @Override void initialize(); }
CopyDataPublisher extends DataPublisher implements UnpublishedHandling { @Override public void publishData(Collection<? extends WorkUnitState> states) throws IOException { Multimap<CopyEntity.DatasetAndPartition, WorkUnitState> datasets = groupByFileSet(states); boolean allDatasetsPublished = true; for (CopyEntity.DatasetAndPartition datasetAndPartition : datasets.keySet()) { try { this.publishFileSet(datasetAndPartition, datasets.get(datasetAndPartition)); } catch (Throwable e) { CopyEventSubmitterHelper.submitFailedDatasetPublish(this.eventSubmitter, datasetAndPartition); log.error("Failed to publish " + datasetAndPartition.getDataset().getDatasetURN(), e); allDatasetsPublished = false; } } if (!allDatasetsPublished) { throw new IOException("Not all datasets published successfully"); } } CopyDataPublisher(State state); @Override boolean isThreadSafe(); @Override void publishData(Collection<? extends WorkUnitState> states); @Override void handleUnpublishedWorkUnits(Collection<? extends WorkUnitState> states); @Override void publishMetadata(Collection<? extends WorkUnitState> states); @Override void close(); @Override void initialize();  }