private void closeTaskDirty(final Task task, final boolean removeFromTasksRegistry) {
        try {
            // we call this function only to flush the case if necessary
            // before suspending and closing the topology
            task.prepareCommit();
        } catch (final RuntimeException swallow) {
            log.error("Error flushing caches of dirty task {}", task.id(), swallow);
        }

        try {
            task.suspend();
        } catch (final RuntimeException swallow) {
            log.error("Error suspending dirty task {}: {}", task.id(), swallow.getMessage());
        }

        task.closeDirty();

        try {
            if (removeFromTasksRegistry) {
                tasks.removeTask(task);
            }

            if (task.isActive()) {
                activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(task.id());
            }
        } catch (final RuntimeException swallow) {
            log.error("Error removing dirty task {}: {}", task.id(), swallow.getMessage());
        }
    }
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