private void cleanBaseDirsAndReadOffsetFiles() {
    LOG.debug("Cleaning base directories for stores.");
    StorageConfig storageConfig = new StorageConfig(config);
    FileUtil fileUtil = new FileUtil();
    taskStores.forEach((storeName, storageEngine) -> {
      if (!storageEngine.getStoreProperties().isLoggedStore()) {
        File nonLoggedStorePartitionDir =
            storageManagerUtil.getTaskStoreDir(nonLoggedStoreBaseDirectory, storeName, taskModel.getTaskName(), taskModel.getTaskMode());
        LOG.info("Got non logged storage partition directory as " + nonLoggedStorePartitionDir.toPath().toString());

        if (nonLoggedStorePartitionDir.exists()) {
          LOG.info("Deleting non logged storage partition directory " + nonLoggedStorePartitionDir.toPath().toString());
          fileUtil.rm(nonLoggedStorePartitionDir);
        }
      } else {
        File loggedStorePartitionDir =
            storageManagerUtil.getTaskStoreDir(loggedStoreBaseDirectory, storeName, taskModel.getTaskName(), taskModel.getTaskMode());
        LOG.info("Got logged storage partition directory as " + loggedStorePartitionDir.toPath().toString());

        // Delete the logged store if it is not valid.
        if (!isLoggedStoreValid(storeName, loggedStorePartitionDir) || storageConfig.cleanLoggedStoreDirsOnStart(storeName)) {
          LOG.info("Deleting logged storage partition directory " + loggedStorePartitionDir.toPath().toString());
          fileUtil.rm(loggedStorePartitionDir);
        } else {

          SystemStreamPartition changelogSSP = new SystemStreamPartition(storeChangelogs.get(storeName), taskModel.getChangelogPartition());
          Map<SystemStreamPartition, String> offset =
              storageManagerUtil.readOffsetFile(loggedStorePartitionDir, Collections.singleton(changelogSSP), false);
          LOG.info("Read offset {} for the store {} from logged storage partition directory {}", offset, storeName, loggedStorePartitionDir);

          if (offset.containsKey(changelogSSP)) {
            fileOffsets.put(changelogSSP, offset.get(changelogSSP));
          }
        }
      }
    });
  }
For each store for this task,
a. Deletes the corresponding non-logged-store base dir.
b. Deletes the logged-store-base-dir (depending on {@param cleanLoggedStoreDirs} value). See {@link #isLoggedStoreValid} for validation semantics.
c. If the logged-store-base-dir is valid, this method reads the offset file and stores each offset.
samza-core/src/main/java/org/apache/samza/storage/NonTransactionalStateTaskRestoreManager.java