OperatorImplGraph { static Multimap<SystemStream, String> getStreamToConsumerTasks(JobModel jobModel) { Multimap<SystemStream, String> streamToConsumerTasks = HashMultimap.create(); jobModel.getContainers().values().forEach(containerModel -> { containerModel.getTasks().values().forEach(taskModel -> { taskModel.getSystemStreamPartitions().forEach(ssp -> { streamToConsumerTasks.put(ssp.getSystemStream(), taskModel.getTaskName().getTaskName()); }); }); }); return streamToConsumerTasks; } OperatorImplGraph(OperatorSpecGraph specGraph, Context context, Clock clock); InputOperatorImpl getInputOperator(SystemStream systemStream); void close(); Collection<InputOperatorImpl> getAllInputOperators();  }
@Test public void testGetStreamToConsumerTasks() { String system = "test-system"; String streamId0 = "test-stream-0"; String streamId1 = "test-stream-1"; HashMap<String, String> configs = new HashMap<>(); configs.put(JobConfig.JOB_NAME, "test-app"); configs.put(JobConfig.JOB_DEFAULT_SYSTEM, "test-system"); StreamTestUtils.addStreamConfigs(configs, streamId0, system, streamId0); StreamTestUtils.addStreamConfigs(configs, streamId1, system, streamId1); Config config = new MapConfig(configs); when(this.context.getJobContext().getConfig()).thenReturn(config); SystemStreamPartition ssp0 = new SystemStreamPartition(system, streamId0, new Partition(0)); SystemStreamPartition ssp1 = new SystemStreamPartition(system, streamId0, new Partition(1)); SystemStreamPartition ssp2 = new SystemStreamPartition(system, streamId1, new Partition(0)); TaskName task0 = new TaskName("Task 0"); TaskName task1 = new TaskName("Task 1"); Set<SystemStreamPartition> ssps = new HashSet<>(); ssps.add(ssp0); ssps.add(ssp2); TaskModel tm0 = new TaskModel(task0, ssps, new Partition(0)); ContainerModel cm0 = new ContainerModel("c0", Collections.singletonMap(task0, tm0)); TaskModel tm1 = new TaskModel(task1, Collections.singleton(ssp1), new Partition(1)); ContainerModel cm1 = new ContainerModel("c1", Collections.singletonMap(task1, tm1)); Map<String, ContainerModel> cms = new HashMap<>(); cms.put(cm0.getId(), cm0); cms.put(cm1.getId(), cm1); JobModel jobModel = new JobModel(config, cms); Multimap<SystemStream, String> streamToTasks = OperatorImplGraph.getStreamToConsumerTasks(jobModel); assertEquals(streamToTasks.get(ssp0.getSystemStream()).size(), 2); assertEquals(streamToTasks.get(ssp2.getSystemStream()).size(), 1); }