private boolean hasIntermediateStreams(OperatorSpecGraph specGraph) {
    return !Collections.disjoint(specGraph.getInputOperators().keySet(), specGraph.getOutputStreams().keySet());
  }
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