private List<TaskDirectory> listTaskDirectories(final FileFilter filter) {
        final List<TaskDirectory> taskDirectories = new ArrayList<>();
        if (hasPersistentStores && stateDir.exists()) {
            if (hasNamedTopologies) {
                for (final File namedTopologyDir : listNamedTopologyDirs()) {
                    final String namedTopology = parseNamedTopologyFromDirectory(namedTopologyDir.getName());
                    final File[] taskDirs = namedTopologyDir.listFiles(filter);
                    if (taskDirs != null) {
                        taskDirectories.addAll(Arrays.stream(taskDirs)
                            .map(f -> new TaskDirectory(f, namedTopology)).collect(Collectors.toList()));
                    }
                }
            } else {
                final File[] taskDirs =
                    stateDir.listFiles(filter);
                if (taskDirs != null) {
                    taskDirectories.addAll(Arrays.stream(taskDirs)
                                               .map(f -> new TaskDirectory(f, null)).collect(Collectors.toList()));
                }
            }
        }

        return taskDirectories;
    }
List all of the task directories along with their parent directory if they belong to a named topology
@return The list of all the existing local directories for stream tasks
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java