OperatorImplGraph { static Map<SystemStream, Integer> getProducerTaskCountForIntermediateStreams( Multimap<SystemStream, String> streamToConsumerTasks, Multimap<SystemStream, SystemStream> intermediateToInputStreams) { Map<SystemStream, Integer> result = new HashMap<>(); intermediateToInputStreams.asMap().entrySet().forEach(entry -> result.put(entry.getKey(), entry.getValue().stream() .flatMap(systemStream -> streamToConsumerTasks.get(systemStream).stream()) .collect(Collectors.toSet()).size())); return result; } OperatorImplGraph(OperatorSpecGraph specGraph, Context context, Clock clock); InputOperatorImpl getInputOperator(SystemStream systemStream); void close(); Collection<InputOperatorImpl> getAllInputOperators();  }
@Test public void testGetProducerTaskCountForIntermediateStreams() { String inputStreamId1 = "input1"; String inputStreamId2 = "input2"; String inputStreamId3 = "input3"; String inputSystem1 = "system1"; String inputSystem2 = "system2"; SystemStream input1 = new SystemStream("system1", "intput1"); SystemStream input2 = new SystemStream("system2", "intput2"); SystemStream input3 = new SystemStream("system2", "intput3"); SystemStream int1 = new SystemStream("system1", "int1"); SystemStream int2 = new SystemStream("system1", "int2"); String task0 = "Task 0"; String task1 = "Task 1"; String task2 = "Task 2"; Multimap<SystemStream, String> streamToConsumerTasks = HashMultimap.create(); streamToConsumerTasks.put(input1, task0); streamToConsumerTasks.put(input1, task1); streamToConsumerTasks.put(input2, task1); streamToConsumerTasks.put(input2, task2); streamToConsumerTasks.put(input3, task1); streamToConsumerTasks.put(int1, task0); streamToConsumerTasks.put(int1, task1); streamToConsumerTasks.put(int2, task0); Multimap<SystemStream, SystemStream> intermediateToInputStreams = HashMultimap.create(); intermediateToInputStreams.put(int1, input1); intermediateToInputStreams.put(int1, input2); intermediateToInputStreams.put(int2, input2); intermediateToInputStreams.put(int2, input3); Map<SystemStream, Integer> counts = OperatorImplGraph.getProducerTaskCountForIntermediateStreams( streamToConsumerTasks, intermediateToInputStreams); assertTrue(counts.get(int1) == 3); assertTrue(counts.get(int2) == 2); }