public static Map<String, TaskRestoreManager> createTaskRestoreManagers(TaskName taskName,
      Map<String, Set<String>> backendFactoryStoreNames,
      Map<String, StateBackendFactory> stateBackendFactories,
      Map<String, StorageEngineFactory<Object, Object>> storageEngineFactories,
      Map<String, SystemConsumer> storeConsumers,
      Map<TaskName, Map<String, StorageEngine>> inMemoryStores,
      SystemAdmins systemAdmins,
      ExecutorService restoreExecutor,
      TaskModel taskModel, JobContext jobContext, ContainerContext containerContext,
      SamzaContainerMetrics samzaContainerMetrics,
      Map<TaskName, TaskInstanceMetrics> taskInstanceMetrics,
      Map<TaskName, TaskInstanceCollector> taskInstanceCollectors,
      Map<String, Serde<Object>> serdes,
      File loggedStoreBaseDirectory, File nonLoggedStoreBaseDirectory,
      Config config, Clock clock) {
    // Get the factories for the task based on the stores of the tasks to be restored from the factory
    Map<String, TaskRestoreManager> backendFactoryRestoreManagers = new HashMap<>(); // backendFactoryName -> restoreManager
    MetricsRegistry taskMetricsRegistry =
        taskInstanceMetrics.get(taskName) != null ?
            taskInstanceMetrics.get(taskName).registry() : new MetricsRegistryMap();

    backendFactoryStoreNames.forEach((factoryName, storeNames) -> {
      StateBackendFactory factory = stateBackendFactories.get(factoryName);
      if (factory == null) {
        throw new SamzaException(
            String.format("Required restore state backend factory: %s not found in configured factories %s",
                factoryName, String.join(", ", stateBackendFactories.keySet())));
      }
      KafkaChangelogRestoreParams kafkaChangelogRestoreParams = new KafkaChangelogRestoreParams(storeConsumers,
          inMemoryStores.get(taskName), systemAdmins.getSystemAdmins(), storageEngineFactories, serdes,
          taskInstanceCollectors.get(taskName));
      TaskRestoreManager restoreManager = factory.getRestoreManager(jobContext, containerContext, taskModel,
          restoreExecutor, taskMetricsRegistry, storeNames, config, clock,
          loggedStoreBaseDirectory, nonLoggedStoreBaseDirectory,
          kafkaChangelogRestoreParams);

      backendFactoryRestoreManagers.put(factoryName, restoreManager);
    });
    samzaContainerMetrics.addStoresRestorationGauge(taskName);
    return backendFactoryRestoreManagers;
  }
Creates SystemConsumers for store restore, one consumer per system.
@return map of system name to its system consumer.
samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManagerUtil.java