public Optional<StorageEngine> getStore(TaskName taskName, String storeName) {
    if (!isStarted) {
      throw new SamzaException(String.format(
          "Attempting to access store %s for task %s before ContainerStorageManager is started.",
          storeName, taskName));
    }
    return Optional.ofNullable(this.taskStores.get(taskName).get(storeName));
  }
Get the {@link StorageEngine} instance with a given name for a given task.
@param taskName the task name for which the storage engine is desired.
@param storeName the desired store's name.
@return the task store.
samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java