static Multimap<SystemStream, String> getStreamToConsumerTasks(JobModel jobModel) {
    Multimap<SystemStream, String> streamToConsumerTasks = HashMultimap.create();
    jobModel.getContainers().values().forEach(containerModel -> {
      containerModel.getTasks().values().forEach(taskModel -> {
        taskModel.getSystemStreamPartitions().forEach(ssp -> {
          streamToConsumerTasks.put(ssp.getSystemStream(), taskModel.getTaskName().getTaskName());
        });
      });
    });
    return streamToConsumerTasks;
  }
calculate the mapping from input streams to consumer tasks
@param jobModel JobModel object
@return mapping from input stream to tasks
samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImplGraph.java