boolean directoryForTaskIsEmpty(final TaskId taskId) {
        final File taskDir = getOrCreateDirectoryForTask(taskId);

        return taskDirIsEmpty(taskDir);
    }
Decide if the directory of the task is empty or not
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java