List<TaskDirectory> listNonEmptyTaskDirectories() {
        return listTaskDirectories(pathname -> {
            if (!pathname.isDirectory() || !TASK_DIR_PATH_NAME.matcher(pathname.getName()).matches()) {
                return false;
            } else {
                return !taskDirIsEmpty(pathname);
            }
        });
    }
List all of the task directories that are non-empty
@return The list of all the non-empty local directories for stream tasks
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java