private void setupBaseDirs() {
    LOG.debug("Setting up base directories for stores.");
    taskStores.forEach((storeName, storageEngine) -> {
      if (storageEngine.getStoreProperties().isLoggedStore()) {

        File loggedStorePartitionDir =
            storageManagerUtil.getTaskStoreDir(loggedStoreBaseDirectory, storeName, taskModel.getTaskName(), taskModel.getTaskMode());

        LOG.info("Using logged storage partition directory: " + loggedStorePartitionDir.toPath().toString()
            + " for store: " + storeName);

        if (!loggedStorePartitionDir.exists()) {
          loggedStorePartitionDir.mkdirs();
        }
      } else {
        File nonLoggedStorePartitionDir =
            storageManagerUtil.getTaskStoreDir(nonLoggedStoreBaseDirectory, storeName, taskModel.getTaskName(), taskModel.getTaskMode());
        LOG.info("Using non logged storage partition directory: " + nonLoggedStorePartitionDir.toPath().toString()
            + " for store: " + storeName);
        nonLoggedStorePartitionDir.mkdirs();
      }
    });
  }
Create stores' base directories for logged-stores if they dont exist.
samza-core/src/main/java/org/apache/samza/storage/NonTransactionalStateTaskRestoreManager.java