@Test(expectedExceptions = IllegalArgumentException.class) public void testBogusMetadataMerger() throws IOException { State s = buildDefaultState(1); s.setProp(ConfigurationKeys.DATA_PUBLISH_WRITER_METADATA_KEY, "true"); s.setProp(ConfigurationKeys.DATA_PUBLISH_WRITER_METADATA_MERGER_NAME_KEY, "java.lang.String"); s.setProp(ConfigurationKeys.DATA_PUBLISHER_METADATA_STR, "foobar"); WorkUnitState wuState = new WorkUnitState(); addStateToWorkunit(s, wuState); BaseDataPublisher publisher = new BaseDataPublisher(s); publisher.publishMetadata(Collections.singletonList(wuState)); }
@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);  }