@Test public void testMergedMetadata() throws IOException { final int numBranches = 2; final int numWorkUnits = 10; State s = buildDefaultState(numBranches); for (int i = 0; i < numBranches; i++) { String mdKeyName = ForkOperatorUtils .getPropertyNameForBranch(ConfigurationKeys.DATA_PUBLISH_WRITER_METADATA_KEY, numBranches, i); String mdMergerKeyName = ForkOperatorUtils .getPropertyNameForBranch(ConfigurationKeys.DATA_PUBLISH_WRITER_METADATA_MERGER_NAME_KEY, numBranches, i); s.setProp(mdKeyName, "true"); s.setProp(mdMergerKeyName, (i % 2) == 0 ? TestAdditionMerger.class.getName() : TestMultiplicationMerger.class.getName()); } List<WorkUnitState> workUnits = new ArrayList<>(); for (int workUnitId = 0; workUnitId < numWorkUnits; workUnitId++) { WorkUnitState wuState = new WorkUnitState(); addStateToWorkunit(s, wuState); for (int branchId = 0; branchId < numBranches; branchId++) { String mdForBranchName = ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.WRITER_METADATA_KEY, numBranches, branchId); wuState.setProp(mdForBranchName, String.valueOf((branchId + 1) * (workUnitId + 1))); } workUnits.add(wuState); } BaseDataPublisher publisher = new BaseDataPublisher(s); publisher.publishMetadata(workUnits); for (int branch = 0; branch < numBranches; branch++) { int expectedSum = (branch % 2 == 0) ? 0 : 1; for (int i = 0; i < numWorkUnits; i++) { if (branch % 2 == 0) { expectedSum += (branch + 1) * (i + 1); } else { expectedSum *= (branch + 1) * (i + 1); } } try (InputStream mdStream = new FileInputStream(openMetadataFile(s, numBranches, branch))) { String mdBytes = IOUtils.toString(mdStream, StandardCharsets.UTF_8); Assert.assertEquals(mdBytes, String.valueOf(expectedSum), "Expected to read back correctly merged metadata from string"); } } }
@Override public void publishMetadata(Collection<? extends WorkUnitState> states) throws IOException { Set<String> partitions = new HashSet<>(); mergeMetadataAndCollectPartitionNames(states, partitions); partitions.removeIf(Objects::isNull); WorkUnitState anyState = states.iterator().next(); for (int branchId = 0; branchId < numBranches; branchId++) { String mdOutputPath = getMetadataOutputPathFromState(anyState, branchId); String userSpecifiedPath = getUserSpecifiedOutputPathFromState(anyState, branchId); if (partitions.isEmpty() || userSpecifiedPath != null) { publishMetadata(getMergedMetadataForPartitionAndBranch(null, branchId), branchId, getMetadataOutputFileForBranch(anyState, branchId)); } else { String metadataFilename = getMetadataFileNameForBranch(anyState, branchId); if (mdOutputPath == null || metadataFilename == null) { LOG.info("Metadata filename not set for branch " + String.valueOf(branchId) + ": not publishing metadata."); continue; } for (String partition : partitions) { publishMetadata(getMergedMetadataForPartitionAndBranch(partition, branchId), branchId, new Path(new Path(mdOutputPath, partition), metadataFilename)); } } } }
BaseDataPublisher extends SingleTaskDataPublisher { @Override public void publishMetadata(Collection<? extends WorkUnitState> states) throws IOException { Set<String> partitions = new HashSet<>(); mergeMetadataAndCollectPartitionNames(states, partitions); partitions.removeIf(Objects::isNull); WorkUnitState anyState = states.iterator().next(); for (int branchId = 0; branchId < numBranches; branchId++) { String mdOutputPath = getMetadataOutputPathFromState(anyState, branchId); String userSpecifiedPath = getUserSpecifiedOutputPathFromState(anyState, branchId); if (partitions.isEmpty() || userSpecifiedPath != null) { publishMetadata(getMergedMetadataForPartitionAndBranch(null, branchId), branchId, getMetadataOutputFileForBranch(anyState, branchId)); } else { String metadataFilename = getMetadataFileNameForBranch(anyState, branchId); if (mdOutputPath == null || metadataFilename == null) { LOG.info("Metadata filename not set for branch " + String.valueOf(branchId) + ": not publishing metadata."); continue; } for (String partition : partitions) { publishMetadata(getMergedMetadataForPartitionAndBranch(partition, branchId), branchId, new Path(new Path(mdOutputPath, partition), metadataFilename)); } } } } }
BaseDataPublisher extends SingleTaskDataPublisher { @Override public void publishMetadata(Collection<? extends WorkUnitState> states) throws IOException { Set<String> partitions = new HashSet<>(); mergeMetadataAndCollectPartitionNames(states, partitions); partitions.removeIf(Objects::isNull); WorkUnitState anyState = states.iterator().next(); for (int branchId = 0; branchId < numBranches; branchId++) { String mdOutputPath = getMetadataOutputPathFromState(anyState, branchId); String userSpecifiedPath = getUserSpecifiedOutputPathFromState(anyState, branchId); if (partitions.isEmpty() || userSpecifiedPath != null) { publishMetadata(getMergedMetadataForPartitionAndBranch(null, branchId), branchId, getMetadataOutputFileForBranch(anyState, branchId)); } else { String metadataFilename = getMetadataFileNameForBranch(anyState, branchId); if (mdOutputPath == null || metadataFilename == null) { LOG.info("Metadata filename not set for branch " + String.valueOf(branchId) + ": not publishing metadata."); continue; } for (String partition : partitions) { publishMetadata(getMergedMetadataForPartitionAndBranch(partition, branchId), branchId, new Path(new Path(mdOutputPath, partition), metadataFilename)); } } } } BaseDataPublisher(State state); }
BaseDataPublisher extends SingleTaskDataPublisher { @Override public void publishMetadata(Collection<? extends WorkUnitState> states) throws IOException { Set<String> partitions = new HashSet<>(); mergeMetadataAndCollectPartitionNames(states, partitions); partitions.removeIf(Objects::isNull); WorkUnitState anyState = states.iterator().next(); for (int branchId = 0; branchId < numBranches; branchId++) { String mdOutputPath = getMetadataOutputPathFromState(anyState, branchId); String userSpecifiedPath = getUserSpecifiedOutputPathFromState(anyState, branchId); if (partitions.isEmpty() || userSpecifiedPath != null) { publishMetadata(getMergedMetadataForPartitionAndBranch(null, branchId), branchId, getMetadataOutputFileForBranch(anyState, branchId)); } else { String metadataFilename = getMetadataFileNameForBranch(anyState, branchId); if (mdOutputPath == null || metadataFilename == null) { LOG.info("Metadata filename not set for branch " + String.valueOf(branchId) + ": not publishing metadata."); continue; } for (String partition : partitions) { publishMetadata(getMergedMetadataForPartitionAndBranch(partition, branchId), branchId, new Path(new Path(mdOutputPath, partition), metadataFilename)); } } } } BaseDataPublisher(State state); @Override void initialize(); @Override void close(); @Override void publishData(WorkUnitState state); @Override void publishData(Collection<? extends WorkUnitState> states); @Override void publishMetadata(Collection<? extends WorkUnitState> states); @Override void publishMetadata(WorkUnitState state); }
BaseDataPublisher extends SingleTaskDataPublisher { @Override public void publishMetadata(Collection<? extends WorkUnitState> states) throws IOException { Set<String> partitions = new HashSet<>(); mergeMetadataAndCollectPartitionNames(states, partitions); partitions.removeIf(Objects::isNull); WorkUnitState anyState = states.iterator().next(); for (int branchId = 0; branchId < numBranches; branchId++) { String mdOutputPath = getMetadataOutputPathFromState(anyState, branchId); String userSpecifiedPath = getUserSpecifiedOutputPathFromState(anyState, branchId); if (partitions.isEmpty() || userSpecifiedPath != null) { publishMetadata(getMergedMetadataForPartitionAndBranch(null, branchId), branchId, getMetadataOutputFileForBranch(anyState, branchId)); } else { String metadataFilename = getMetadataFileNameForBranch(anyState, branchId); if (mdOutputPath == null || metadataFilename == null) { LOG.info("Metadata filename not set for branch " + String.valueOf(branchId) + ": not publishing metadata."); continue; } for (String partition : partitions) { publishMetadata(getMergedMetadataForPartitionAndBranch(partition, branchId), branchId, new Path(new Path(mdOutputPath, partition), metadataFilename)); } } } } BaseDataPublisher(State state); @Override void initialize(); @Override void close(); @Override void publishData(WorkUnitState state); @Override void publishData(Collection<? extends WorkUnitState> states); @Override void publishMetadata(Collection<? extends WorkUnitState> states); @Override void publishMetadata(WorkUnitState state);  }