private List<File> listNamedTopologyDirs() {
        final File[] namedTopologyDirectories = stateDir.listFiles(f -> f.getName().startsWith("__") &&  f.getName().endsWith("__"));
        return namedTopologyDirectories != null ? Arrays.asList(namedTopologyDirectories) : Collections.emptyList();
    }
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