public static Map<String, SystemStream> getActiveTaskChangelogSystemStreams(
      Map<String, SystemStream> changelogSystemStreams, ContainerModel containerModel) {
    if (MapUtils.invertMap(changelogSystemStreams).size() != changelogSystemStreams.size()) {
      throw new SamzaException("Two stores cannot have the same changelog system-stream");
    }

    Map<SystemStreamPartition, String> changelogSSPToStore = new HashMap<>();
    changelogSystemStreams.forEach((storeName, systemStream) ->
        containerModel.getTasks().forEach((taskName, taskModel) ->
            changelogSSPToStore.put(new SystemStreamPartition(systemStream, taskModel.getChangelogPartition()), storeName))
    );

    getTasks(containerModel, TaskMode.Standby).forEach((taskName, taskModel) -> {
      changelogSystemStreams.forEach((storeName, systemStream) -> {
        SystemStreamPartition ssp = new SystemStreamPartition(systemStream, taskModel.getChangelogPartition());
        changelogSSPToStore.remove(ssp);
      });
    });

    // changelogSystemStreams correspond only to active tasks (since those of standby-tasks moved to sideInputs above)
    return MapUtils.invertMap(changelogSSPToStore).entrySet().stream()
        .collect(Collectors.toMap(Map.Entry::getKey, x -> x.getValue().getSystemStream()));
  }
Returns a map of storeNames to changelogSSPs associated with the active tasks.
The standby changelogs will be consumed and restored as side inputs.

@param changelogSystemStreams the passed in map of storeName to changelogSystemStreams
@param containerModel the container's model
@return A map of storeName to changelogSSPs across all active tasks, assuming no two stores have the same changelogSSP
samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManagerUtil.java