public static Map<TaskName, Map<String, StorageEngine>> createTaskStores(Set<String> storesToCreate,
      Map<String, StorageEngineFactory<Object, Object>> storageEngineFactories,
      Set<String> sideInputStoreNames,
      Map<String, SystemStream> activeTaskChangelogSystemStreams,
      ContainerModel containerModel, JobContext jobContext, ContainerContext containerContext,
      Map<String, Serde<Object>> serdes,
      Map<TaskName, TaskInstanceMetrics> taskInstanceMetrics,
      Map<TaskName, TaskInstanceCollector> taskInstanceCollectors,
      StorageManagerUtil storageManagerUtil,
      File loggedStoreBaseDirectory, File nonLoggedStoreBaseDirectory,
      Config config) {
    Map<TaskName, Map<String, StorageEngine>> taskStores = new HashMap<>();

    // iterate over each task and each storeName
    for (Map.Entry<TaskName, TaskModel> task : containerModel.getTasks().entrySet()) {
      TaskName taskName = task.getKey();
      TaskModel taskModel = task.getValue();
      if (!taskStores.containsKey(taskName)) {
        taskStores.put(taskName, new HashMap<>());
      }

      for (String storeName : storesToCreate) {
        File storeDirectory = getStoreDirPath(storeName, config, activeTaskChangelogSystemStreams,
            sideInputStoreNames, taskName, taskModel, storageManagerUtil, loggedStoreBaseDirectory, nonLoggedStoreBaseDirectory);

        // if taskInstanceMetrics are specified use those for store metrics,
        // otherwise (in case of StorageRecovery) use a blank MetricsRegistryMap
        MetricsRegistry storeMetricsRegistry =
            taskInstanceMetrics.get(taskName) != null ?
                taskInstanceMetrics.get(taskName).registry() : new MetricsRegistryMap();

        StorageEngine storageEngine =
            createStore(storeName, storeDirectory, StorageEngineFactory.StoreMode.ReadWrite,
                storageEngineFactories, activeTaskChangelogSystemStreams,
                taskModel, jobContext, containerContext,
                serdes, storeMetricsRegistry, taskInstanceCollectors.get(taskName),
                config);

        // add created store to map
        taskStores.get(taskName).put(storeName, storageEngine);

        LOG.info("Created task store {} in read-write mode for task {} in path {}",
            storeName, taskName, storeDirectory.getAbsolutePath());
      }
    }
    return taskStores;
  }
Create taskStores for all stores in storesToCreate.
The store mode is chosen as read-write mode.
Adds newly created stores to storeDirectoryPaths.
samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManagerUtil.java