public static Map<String, SystemConsumer> createStoreChangelogConsumers(
      Map<String, SystemStream> activeTaskChangelogSystemStreams,
      Map<String, SystemFactory> systemFactories, MetricsRegistry registry, Config config) {
    Set<String> activeTaskChangelogSystems = activeTaskChangelogSystemStreams.values().stream()
        .map(SystemStream::getSystem)
        .collect(Collectors.toSet());

    Map<String, SystemConsumer> systemNameToSystemConsumers = ContainerStorageManagerUtil.createSystemConsumers(
        activeTaskChangelogSystems, systemFactories, registry, config);

    // Map of each storeName to its respective systemConsumer
    Map<String, SystemConsumer> storeConsumers = new HashMap<>();

    // Populate the map of storeName to its relevant systemConsumer
    for (String storeName : activeTaskChangelogSystemStreams.keySet()) {
      storeConsumers.put(storeName,
          systemNameToSystemConsumers.get(activeTaskChangelogSystemStreams.get(storeName).getSystem()));
    }
    return storeConsumers;
  }
Returns a map from store names to their corresponding changelog system consumers.
@return map of store name to its changelog system consumer.
samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManagerUtil.java