private void closeTaskClean(final Task task) {
        task.closeClean();
        tasks.removeTask(task);
        if (task.isActive()) {
            activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(task.id());
        }
    }
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