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;
        }
    }
Get the lock for the {@link TaskId}s directory if it is available
@param taskId task id
@return true if successful
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java