void closeAndCleanUpTasks(final Collection<Task> activeTasks, final Collection<Task> standbyTasks, final boolean clean) {
        final AtomicReference<RuntimeException> firstException = new AtomicReference<>(null);

        final Set<TaskId> ids =
            activeTasks.stream()
                .map(Task::id)
                .collect(Collectors.toSet());
        maybeLockTasks(ids);

        final Set<Task> tasksToCloseDirty = new HashSet<>();
        tasksToCloseDirty.addAll(tryCloseCleanActiveTasks(activeTasks, clean, firstException));
        tasksToCloseDirty.addAll(tryCloseCleanStandbyTasks(standbyTasks, clean, firstException));

        for (final Task task : tasksToCloseDirty) {
            closeTaskDirty(task, true);
        }

        maybeUnlockTasks(ids);

        final RuntimeException exception = firstException.get();
        if (exception != null) {
            throw exception;
        }
    }
Closes and cleans up after the provided tasks, including closing their corresponding task producers
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java