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;
    }
@Test
    public void shouldThrowProcessorStateExceptionIfStateDirOccupied() throws IOException {
        final TaskId taskId = new TaskId(0, 0);

        // Replace application's stateDir to regular file
        Utils.delete(appDir);
        Files.createFile(appDir.toPath());

        assertThrows(ProcessorStateException.class, () -> directory.getOrCreateDirectoryForTask(taskId));
    }