synchronized void unlock(final TaskId taskId) {
        final Thread lockOwner = lockedTasksToOwner.get(taskId);
        if (lockOwner != null && lockOwner.equals(Thread.currentThread())) {
            lockedTasksToOwner.remove(taskId);
            log.debug("{} Released state dir lock for task {}", logPrefix(), taskId);
        }
    }
@Test
    public void shouldBeAbleToUnlockEvenWithoutLocking() {
        final TaskId taskId = new TaskId(0, 0);
        directory.unlock(taskId);
    }