@VisibleForTesting
  static Map<TaskName, Map<String, Set<SystemStreamPartition>>> getTaskSideInputSSPs(
      Map<String, Set<SystemStream>> sideInputSystemStreams,
      Map<String, SystemStream> changelogSystemStreams,
      ContainerModel containerModel) {
    Map<TaskName, Map<String, Set<SystemStreamPartition>>> taskSideInputSSPs = new HashMap<>();

    containerModel.getTasks().forEach((taskName, taskModel) -> {
      taskSideInputSSPs.putIfAbsent(taskName, new HashMap<>());
      sideInputSystemStreams.keySet().forEach(storeName -> {
        Set<SystemStreamPartition> taskSideInputs = taskModel.getSystemStreamPartitions().stream()
            .filter(ssp -> sideInputSystemStreams.get(storeName).contains(ssp.getSystemStream()))
            .collect(Collectors.toSet());
        taskSideInputSSPs.get(taskName).put(storeName, taskSideInputs);
      });
    });

    ContainerStorageManagerUtil.getTasks(containerModel, TaskMode.Standby).forEach((taskName, taskModel) -> {
      taskSideInputSSPs.putIfAbsent(taskName, new HashMap<>());
      changelogSystemStreams.forEach((storeName, systemStream) -> {
        SystemStreamPartition ssp = new SystemStreamPartition(systemStream, taskModel.getChangelogPartition());
        taskSideInputSSPs.get(taskName).put(storeName, Collections.singleton(ssp));
      });
    });

    return taskSideInputSSPs;
  }
Add all sideInputs to a map of maps, indexed first by taskName, then by sideInput store name.

@param sideInputSystemStreams the map of store to sideInput system stream
@param changelogSystemStreams the map of store to changelog system stream
@param containerModel the containerModel to use
@return taskSideInputSSPs map
samza-core/src/main/scala/org/apache/samza/storage/SideInputsManager.java