public List<File> getTaskStoreCheckpointDirs(File storeBaseDir, String storeName,
      TaskName taskName, TaskMode taskMode) {
    try {
      File storeDir = new File(storeBaseDir, storeName);
      String taskStoreName = getTaskStoreDir(storeBaseDir, storeName, taskName, taskMode).getName();

      if (storeDir.exists()) { // new store or no local state
        List<File> checkpointDirs = Files.list(storeDir.toPath())
            .map(Path::toFile)
            .filter(file -> file.getName().contains(taskStoreName + "-"))
            .collect(Collectors.toList());
        return checkpointDirs;
      } else {
        return Collections.emptyList();
      }
    } catch (IOException e) {
      throw new SamzaException(
          String.format("Error finding checkpoint dirs for task: %s mode: %s store: %s in dir: %s",
              taskName, taskMode, storeName, storeBaseDir), e);
    }
  }
Creates and returns a File pointing to the directory for the given store and task, given a particular base directory.
In case of a standby task (TaskMode.Standby), the storeDirectory is the same as it would be for an active task.

@param storeBaseDir the base directory to use
@param storeName the store name to use
@param taskName the task name which is referencing the store
@param taskMode the mode of the given task
@return the partition directory for the store
samza-core/src/main/java/org/apache/samza/storage/StorageManagerUtil.java