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);
        }
    }
Unlock the state directory for the given {@link TaskId}.
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java