private void releaseLockedUnassignedTaskDirectories() {
        final Iterator<TaskId> taskIdIterator = lockedTaskDirectories.iterator();
        final Map<TaskId, Task> allTasks = allTasks();
        while (taskIdIterator.hasNext()) {
            final TaskId id = taskIdIterator.next();
            if (!allTasks.containsKey(id)) {
                stateDirectory.unlock(id);
                taskIdIterator.remove();
            }
        }
    }
We must release the lock for any unassigned tasks that we temporarily locked in preparation for a
rebalance in {@link #tryToLockAllNonEmptyTaskDirectories()}.
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java