public void shutdown() {
    // stop all non side input stores including persistent and non-persistent stores
    if (taskStores != null) {
      this.containerModel.getTasks()
          .forEach((taskName, taskModel) -> taskStores.get(taskName)
              .entrySet().stream()
              .filter(e -> !sideInputStoreNames.contains(e.getKey()))
              .forEach(e -> e.getValue().stop()));
    }

    this.sideInputsManager.shutdown();
    LOG.info("Shutdown complete");
  }
Set of directory paths for all stores restored by this {@link ContainerStorageManager}.
@return the set of all store directory paths
samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java