private void releaseLockedDirectoriesForTasks(final Set<TaskId> tasksToUnlock) {
        final Iterator<TaskId> taskIdIterator = lockedTaskDirectories.iterator();
        while (taskIdIterator.hasNext()) {
            final TaskId id = taskIdIterator.next();
            if (tasksToUnlock.contains(id)) {
                stateDirectory.unlock(id);
                taskIdIterator.remove();
            }
        }
    }
Clean up after closed or removed tasks by making sure to unlock any remaining locked directories for them, for
example unassigned tasks or those in the CREATED state when closed, since Task#close will not unlock them
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java