OperatorImplGraph { static Multimap<SystemStream, SystemStream> getIntermediateToInputStreamsMap( OperatorSpecGraph specGraph, StreamConfig streamConfig) { Multimap<SystemStream, SystemStream> outputToInputStreams = HashMultimap.create(); specGraph.getInputOperators().entrySet().stream() .forEach(entry -> { SystemStream systemStream = streamConfig.streamIdToSystemStream(entry.getKey()); computeOutputToInput(systemStream, entry.getValue(), outputToInputStreams, streamConfig); }); return outputToInputStreams; } OperatorImplGraph(OperatorSpecGraph specGraph, Context context, Clock clock); InputOperatorImpl getInputOperator(SystemStream systemStream); void close(); Collection<InputOperatorImpl> getAllInputOperators();  }
@Test public void testGetOutputToInputStreams() { String inputStreamId1 = "input1"; String inputStreamId2 = "input2"; String inputStreamId3 = "input3"; String inputSystem = "input-system"; String outputStreamId1 = "output1"; String outputStreamId2 = "output2"; String outputSystem = "output-system"; String intStreamId1 = "test-app-1-partition_by-p1"; String intStreamId2 = "test-app-1-partition_by-p2"; String intSystem = "test-system"; HashMap<String, String> configs = new HashMap<>(); configs.put(JobConfig.JOB_NAME, "test-app"); configs.put(JobConfig.JOB_DEFAULT_SYSTEM, intSystem); StreamTestUtils.addStreamConfigs(configs, inputStreamId1, inputSystem, inputStreamId1); StreamTestUtils.addStreamConfigs(configs, inputStreamId2, inputSystem, inputStreamId2); StreamTestUtils.addStreamConfigs(configs, inputStreamId3, inputSystem, inputStreamId3); StreamTestUtils.addStreamConfigs(configs, outputStreamId1, outputSystem, outputStreamId1); StreamTestUtils.addStreamConfigs(configs, outputStreamId2, outputSystem, outputStreamId2); Config config = new MapConfig(configs); when(this.context.getJobContext().getConfig()).thenReturn(config); StreamApplicationDescriptorImpl graphSpec = new StreamApplicationDescriptorImpl(appDesc -> { GenericSystemDescriptor isd = new GenericSystemDescriptor(inputSystem, "mockFactoryClass"); GenericInputDescriptor inputDescriptor1 = isd.getInputDescriptor(inputStreamId1, mock(Serde.class)); GenericInputDescriptor inputDescriptor2 = isd.getInputDescriptor(inputStreamId2, mock(Serde.class)); GenericInputDescriptor inputDescriptor3 = isd.getInputDescriptor(inputStreamId3, mock(Serde.class)); GenericSystemDescriptor osd = new GenericSystemDescriptor(outputSystem, "mockFactoryClass"); GenericOutputDescriptor outputDescriptor1 = osd.getOutputDescriptor(outputStreamId1, mock(Serde.class)); GenericOutputDescriptor outputDescriptor2 = osd.getOutputDescriptor(outputStreamId2, mock(Serde.class)); MessageStream messageStream1 = appDesc.getInputStream(inputDescriptor1).map(m -> m); MessageStream messageStream2 = appDesc.getInputStream(inputDescriptor2).filter(m -> true); MessageStream messageStream3 = appDesc.getInputStream(inputDescriptor3) .filter(m -> true) .partitionBy(m -> "m", m -> m, mock(KVSerde.class), "p1") .map(m -> m); OutputStream<Object> outputStream1 = appDesc.getOutputStream(outputDescriptor1); OutputStream<Object> outputStream2 = appDesc.getOutputStream(outputDescriptor2); messageStream1 .join(messageStream2, mock(JoinFunction.class), mock(Serde.class), mock(Serde.class), mock(Serde.class), Duration.ofHours(2), "j1") .partitionBy(m -> "m", m -> m, mock(KVSerde.class), "p2") .sendTo(outputStream1); messageStream3 .join(messageStream2, mock(JoinFunction.class), mock(Serde.class), mock(Serde.class), mock(Serde.class), Duration.ofHours(1), "j2") .sendTo(outputStream2); }, config); Multimap<SystemStream, SystemStream> outputToInput = OperatorImplGraph.getIntermediateToInputStreamsMap(graphSpec.getOperatorSpecGraph(), new StreamConfig(config)); Collection<SystemStream> inputs = outputToInput.get(new SystemStream(intSystem, intStreamId2)); assertEquals(inputs.size(), 2); assertTrue(inputs.contains(new SystemStream(inputSystem, inputStreamId1))); assertTrue(inputs.contains(new SystemStream(inputSystem, inputStreamId2))); inputs = outputToInput.get(new SystemStream(intSystem, intStreamId1)); assertEquals(inputs.size(), 1); assertEquals(inputs.iterator().next(), new SystemStream(inputSystem, inputStreamId3)); }