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;
  }
calculate the mapping from output streams to input streams
@param specGraph the user {@link OperatorSpecGraph}
@return mapping from output streams to input streams
samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImplGraph.java