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;
  }
calculate the task count that produces to each intermediate streams
@param streamToConsumerTasks input streams to task mapping
@param intermediateToInputStreams intermediate stream to input streams mapping
@return mapping from intermediate stream to task count
samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImplGraph.java