private static void computeOutputToInput(SystemStream input, OperatorSpec opSpec,
                                           Multimap<SystemStream, SystemStream> outputToInputStreams, StreamConfig streamConfig) {
    if (opSpec instanceof PartitionByOperatorSpec) {
      PartitionByOperatorSpec spec = (PartitionByOperatorSpec) opSpec;
      SystemStream systemStream = streamConfig.streamIdToSystemStream(spec.getOutputStream().getStreamId());
      outputToInputStreams.put(systemStream, input);
    } else if (opSpec instanceof BroadcastOperatorSpec) {
      BroadcastOperatorSpec spec = (BroadcastOperatorSpec) opSpec;
      SystemStream systemStream = streamConfig.streamIdToSystemStream(spec.getOutputStream().getStreamId());
      outputToInputStreams.put(systemStream, input);
    } else {
      Collection<OperatorSpec> nextOperators = opSpec.getRegisteredOperatorSpecs();
      nextOperators.forEach(spec -> computeOutputToInput(input, spec, outputToInputStreams, streamConfig));
    }
  }
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