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

    Set<TaskModel> standbyTaskModels = containerModel.getTasks().values().stream()
        .filter(taskModel -> taskModel.getTaskMode().equals(TaskMode.Standby))
        .collect(Collectors.toSet());

    // remove all standby task changelog ssps
    standbyTaskModels.forEach((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()));
  }
Shared cache across KafkaRestoreManagers for the Kafka partition

@param admins system admins used the fetch the stream metadata
@param clock for cache invalidation
@param ssps SSPs to prefetch
@return SSPMetadataCache containing the partition metadata
samza-core/src/main/java/org/apache/samza/storage/KafkaChangelogStateBackendFactory.java