synchronized boolean lock(final TaskId taskId) {
        if (!hasPersistentStores) {
            return true;
        }

        final Thread lockOwner = lockedTasksToOwner.get(taskId);
        if (lockOwner != null) {
            if (lockOwner.equals(Thread.currentThread())) {
                log.trace("{} Found cached state dir lock for task {}", logPrefix(), taskId);
                // we already own the lock
                return true;
            } else {
                // another thread owns the lock
                return false;
            }
        } else if (!stateDir.exists()) {
            log.error("Tried to lock task directory for {} but the state directory does not exist", taskId);
            throw new IllegalStateException("The state directory has been deleted");
        } else {
            lockedTasksToOwner.put(taskId, Thread.currentThread());
            // make sure the task directory actually exists, and create it if not
            getOrCreateDirectoryForTask(taskId);
            return true;
        }
    }
@Test
    public void shouldNotLockStateDirLockedByAnotherThread() throws Exception {
        final TaskId taskId = new TaskId(0, 0);
        final Thread thread = new Thread(() -> directory.lock(taskId));
        thread.start();
        thread.join(30000);
        assertFalse(directory.lock(taskId));
    }