public static Map<TaskName, Map<String, StorageEngine>> createInMemoryStores(
      Map<String, SystemStream> activeTaskChangelogSystemStreams,
      Map<String, StorageEngineFactory<Object, Object>> storageEngineFactories,
      Set<String> sideInputStoreNames,
      ContainerModel containerModel, JobContext jobContext, ContainerContext containerContext,
      Map<TaskName, TaskInstanceMetrics> taskInstanceMetrics,
      Map<TaskName, TaskInstanceCollector> taskInstanceCollectors,
      Map<String, Serde<Object>> serdes,
      StorageManagerUtil storageManagerUtil,
      File loggedStoreBaseDirectory, File nonLoggedStoreBaseDirectory,
      Config config) {
    Set<String> inMemoryStoreNames = getInMemoryStoreNames(storageEngineFactories, config);
    return ContainerStorageManagerUtil.createTaskStores(
        inMemoryStoreNames, storageEngineFactories, sideInputStoreNames,
        activeTaskChangelogSystemStreams, containerModel, jobContext, containerContext, serdes,
        taskInstanceMetrics, taskInstanceCollectors, storageManagerUtil,
        loggedStoreBaseDirectory, nonLoggedStoreBaseDirectory, config);
  }
Method to instantiate a StorageEngine with the given parameters, and populate the storeDirectory paths (used to monitor
disk space).
samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManagerUtil.java