public File getOrCreateDirectoryForTask(final TaskId taskId) {
        final File taskParentDir = getTaskDirectoryParentName(taskId);
        final File taskDir = new File(taskParentDir, StateManagerUtil.toTaskDirString(taskId));
        if (hasPersistentStores) {
            if (!taskDir.exists()) {
                synchronized (taskDirCreationLock) {
                    // to avoid a race condition, we need to check again if the directory does not exist:
                    // otherwise, two threads might pass the outer `if` (and enter the `then` block),
                    // one blocks on `synchronized` while the other creates the directory,
                    // and the blocking one fails when trying to create it after it's unblocked
                    if (!taskParentDir.exists() && !taskParentDir.mkdir()) {
                        throw new ProcessorStateException(
                            String.format("Parent [%s] of task directory [%s] doesn't exist and couldn't be created",
                                taskParentDir.getPath(), taskDir.getPath()));
                    }
                    if (!taskDir.exists() && !taskDir.mkdir()) {
                        throw new ProcessorStateException(
                            String.format("task directory [%s] doesn't exist and couldn't be created", taskDir.getPath()));
                    }
                }
            } else if (!taskDir.isDirectory()) {
                throw new ProcessorStateException(
                    String.format("state directory [%s] can't be created as there is an existing file with the same name", taskDir.getPath()));
            }
        }
        return taskDir;
    }
Get or create the directory for the provided {@link TaskId}.
@return directory for the {@link TaskId}
@throws ProcessorStateException if the task directory does not exist and could not be created
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java