private Map<String, StorageEngine> createStoreEngines(Set<String> storeNames, JobContext jobContext,
      ContainerContext containerContext, Map<String, StorageEngineFactory<Object, Object>> storageEngineFactories,
      Map<String, Serde<Object>> serdes, MetricsRegistry metricsRegistry,
      MessageCollector messageCollector, Map<String, StorageEngine> nonPersistedStores) {
    Map<String, StorageEngine> storageEngines = new HashMap<>();
    // Put non persisted stores
    nonPersistedStores.forEach(storageEngines::put);
    // Create persisted stores
    storeNames.stream().filter(s -> !nonPersistedStores.containsKey(s)).forEach(storeName -> {
      boolean isLogged = this.storeChangelogs.containsKey(storeName);
      File storeBaseDir = isLogged ? this.loggedStoreBaseDirectory : this.nonLoggedStoreBaseDirectory;
      File storeDirectory = storageManagerUtil.getTaskStoreDir(storeBaseDir, storeName, taskModel.getTaskName(),
          taskModel.getTaskMode());
      StorageEngine engine = ContainerStorageManagerUtil.createStore(storeName, storeDirectory,
          StorageEngineFactory.StoreMode.BulkLoad, storageEngineFactories, this.storeChangelogs,
          taskModel, jobContext, containerContext, serdes,
          metricsRegistry, messageCollector, this.config);
      storageEngines.put(storeName, engine);
    });
    return storageEngines;
  }
Returns the offset with which the changelog consumer should be initialized for the given SystemStreamPartition.

If a file offset exists, it represents the last changelog offset which is also reflected in the on-disk state.
In that case, we use the next offset after the file offset, as long as it is newer than the oldest offset
currently available in the stream.

If there isn't a file offset or it's older than the oldest available offset, we simply start with the oldest.

@param systemStreamPartition  the changelog partition for which the offset is needed.
@param systemAdmin                  the [[SystemAdmin]] for the changelog.
@return the offset to from which the changelog consumer should be initialized.
samza-core/src/main/java/org/apache/samza/storage/NonTransactionalStateTaskRestoreManager.java