@VisibleForTesting
  static void setupStoreDirs(
      TaskModel taskModel,
      Map<String, StorageEngine> storeEngines,
      StoreActions storeActions,
      StorageManagerUtil storageManagerUtil,
      FileUtil fileUtil,
      File loggedStoreBaseDirectory,
      File nonLoggedStoreBaseDirectory) {
    TaskName taskName = taskModel.getTaskName();
    TaskMode taskMode = taskModel.getTaskMode();
    ListMultimap<String, File> storeDirsToDelete = storeActions.storeDirsToDelete;
    Map<String, File> storeDirsToRetain = storeActions.storeDirsToRetain;

    // delete all persistent store directories marked for deletion
    storeDirsToDelete.entries().forEach(entry -> {
      String storeName = entry.getKey();
      File storeDirToDelete = entry.getValue();
      LOG.info("Deleting persistent store directory: {} for store: {} in task: {}",
          storeDirToDelete, storeName, taskName);
      fileUtil.rm(storeDirToDelete);
    });

    // rename all retained persistent logged store checkpoint directories to current directory
    storeDirsToRetain.forEach((storeName, storeDirToRetain) -> {
      File currentDir = storageManagerUtil.getTaskStoreDir(
          loggedStoreBaseDirectory, storeName, taskName, taskMode);
      LOG.info("Moving logged store checkpoint directory: {} for store: {} in task: {} to current directory: {}",
          storeDirsToRetain.toString(), storeName, taskName, currentDir);
      storageManagerUtil.restoreCheckpointFiles(storeDirToRetain, currentDir);
      // do not remove the checkpoint directory yet. in case commit fails and container restarts,
      // we can retry the move. if we delete the checkpoint, the current dir will be deleted as well on
      // restart, and we will have to do a full restore.
    });

    // create any missing (not retained) current directories for persistent stores
    storeEngines.forEach((storeName, storageEngine) -> {
      if (storageEngine.getStoreProperties().isPersistedToDisk()) {
        File currentDir;
        if (storageEngine.getStoreProperties().isLoggedStore()) {
          currentDir = storageManagerUtil.getTaskStoreDir(
              loggedStoreBaseDirectory, storeName, taskName, taskMode);
        } else {
          currentDir = storageManagerUtil.getTaskStoreDir(
              nonLoggedStoreBaseDirectory, storeName, taskName, taskMode);
        }

        try {
          if (!fileUtil.exists(currentDir.toPath())) {
            LOG.info("Creating missing persistent store current directory: {} for store: {} in task: {}",
                currentDir, storeName, taskName);
            fileUtil.createDirectories(currentDir.toPath());
          }
        } catch (Exception e) {
          throw new SamzaException(String.format("Error setting up current directory for store: %s", storeName), e);
        }
      }
    });
  }
For each store for this task,
a. Deletes current directory if persistent but non-logged store.
b. Deletes current and checkpoint directories if persistent logged store and directory is marked for deletion
c. Moves the valid persistent logged store checkpoint directory to current directory if marked for retention.
d. Creates all missing (i.e. not retained in step c) persistent logged store dirs.

When this method returns,
a. There will be a empty current dir for each persistent but non-logged store.
b. There will be a current dir for each persistent logged store. This dir may or may not be empty.
c. There will be no remaining checkpoint dirs for persistent logged stores.
samza-core/src/main/java/org/apache/samza/storage/TransactionalStateTaskRestoreManager.java