@Test public void testPublishMultiTasks() throws IOException { WorkUnitState state1 = buildTaskState(2); WorkUnitState state2 = buildTaskState(2); LineageInfo lineageInfo = LineageInfo.getLineageInfo(state1.getTaskBroker()).get(); DatasetDescriptor source = new DatasetDescriptor("kafka", "testTopic"); lineageInfo.setSource(source, state1); lineageInfo.setSource(source, state2); BaseDataPublisher publisher = new BaseDataPublisher(state1); publisher.publishData(ImmutableList.of(state1, state2)); Assert.assertTrue(state1.contains("gobblin.event.lineage.branch.0.destination")); Assert.assertTrue(state1.contains("gobblin.event.lineage.branch.1.destination")); Assert.assertTrue(state2.contains("gobblin.event.lineage.branch.0.destination")); Assert.assertTrue(state2.contains("gobblin.event.lineage.branch.1.destination")); }
@Override public void publishData(WorkUnitState state) throws IOException { for (int branchId = 0; branchId < this.numBranches; branchId++) { publishSingleTaskData(state, branchId); } this.parallelRunnerCloser.close(); }
BaseDataPublisher extends SingleTaskDataPublisher { @Override public void publishData(WorkUnitState state) throws IOException { for (int branchId = 0; branchId < this.numBranches; branchId++) { publishSingleTaskData(state, branchId); } this.parallelRunnerCloser.close(); } }
BaseDataPublisher extends SingleTaskDataPublisher { @Override public void publishData(WorkUnitState state) throws IOException { for (int branchId = 0; branchId < this.numBranches; branchId++) { publishSingleTaskData(state, branchId); } this.parallelRunnerCloser.close(); } BaseDataPublisher(State state); }
BaseDataPublisher extends SingleTaskDataPublisher { @Override public void publishData(WorkUnitState state) throws IOException { for (int branchId = 0; branchId < this.numBranches; branchId++) { publishSingleTaskData(state, branchId); } this.parallelRunnerCloser.close(); } 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 publishData(WorkUnitState state) throws IOException { for (int branchId = 0; branchId < this.numBranches; branchId++) { publishSingleTaskData(state, branchId); } this.parallelRunnerCloser.close(); } 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);  }