@Test public void testPublishSingleTask() throws IOException { WorkUnitState state = buildTaskState(1); LineageInfo lineageInfo = LineageInfo.getLineageInfo(state.getTaskBroker()).get(); DatasetDescriptor source = new DatasetDescriptor("kafka", "testTopic"); lineageInfo.setSource(source, state); BaseDataPublisher publisher = new BaseDataPublisher(state); publisher.publishData(state); Assert.assertTrue(state.contains("gobblin.event.lineage.branch.0.destination")); Assert.assertFalse(state.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);  }