@SuppressWarnings("rawtypes")
  private void getContainerStorageManagers() {
    Set<String> factoryClasses = new StorageConfig(jobConfig).getRestoreFactories();
    Map<String, StateBackendFactory> stateBackendFactories = factoryClasses.stream().collect(
        Collectors.toMap(factoryClass -> factoryClass,
          factoryClass -> ReflectionUtil.getObj(factoryClass, StateBackendFactory.class)));
    Clock clock = SystemClock.instance();
    StreamMetadataCache streamMetadataCache = new StreamMetadataCache(systemAdmins, 5000, clock);
    // don't worry about prefetching for this; looks like the tool doesn't flush to offset files anyways
    Map<String, SystemFactory> systemFactories = new SystemConfig(jobConfig).getSystemFactories();
    CheckpointManager checkpointManager = new TaskConfig(jobConfig)
        .getCheckpointManager(new MetricsRegistryMap()).orElse(null);

    for (ContainerModel containerModel : containers.values()) {
      ContainerContext containerContext = new ContainerContextImpl(containerModel, new MetricsRegistryMap(), null);

      ContainerStorageManager containerStorageManager =
          new ContainerStorageManager(
              checkpointManager,
              containerModel,
              streamMetadataCache,
              systemAdmins,
              changeLogSystemStreams,
              new HashMap<>(),
              storageEngineFactories,
              systemFactories,
              this.getSerdes(),
              jobConfig,
              new HashMap<>(),
              new SamzaContainerMetrics(containerModel.getId(), new MetricsRegistryMap(), ""),
              JobContextImpl.fromConfigWithDefaults(jobConfig, jobModel),
              containerContext,
              stateBackendFactories,
              new HashMap<>(),
              storeBaseDir,
              storeBaseDir,
              null,
              SystemClock.instance());
      this.containerStorageManagers.put(containerModel.getId(), containerStorageManager);
    }
  }
create one TaskStorageManager for each task. Add all of them to the
List<TaskStorageManager>
samza-core/src/main/java/org/apache/samza/storage/StorageRecovery.java