private Collection<Task> tryCloseCleanActiveTasks(final Collection<Task> activeTasksToClose,
                                                      final boolean clean,
                                                      final AtomicReference<RuntimeException> firstException) {
        if (!clean) {
            return activeTaskIterable();
        }
        final Comparator<Task> byId = Comparator.comparing(Task::id);
        final Set<Task> tasksToCommit = new TreeSet<>(byId);
        final Set<Task> tasksToCloseDirty = new TreeSet<>(byId);
        final Set<Task> tasksToCloseClean = new TreeSet<>(byId);
        final Map<Task, Map<TopicPartition, OffsetAndMetadata>> consumedOffsetsAndMetadataPerTask = new HashMap<>();

        // first committing all tasks and then suspend and close them clean
        for (final Task task : activeTasksToClose) {
            try {
                final Map<TopicPartition, OffsetAndMetadata> committableOffsets = task.prepareCommit();
                tasksToCommit.add(task);
                if (!committableOffsets.isEmpty()) {
                    consumedOffsetsAndMetadataPerTask.put(task, committableOffsets);
                }
                tasksToCloseClean.add(task);
            } catch (final TaskMigratedException e) {
                // just ignore the exception as it doesn't matter during shutdown
                tasksToCloseDirty.add(task);
            } catch (final StreamsException e) {
                e.setTaskId(task.id());
                firstException.compareAndSet(null, e);
                tasksToCloseDirty.add(task);
            } catch (final RuntimeException e) {
                firstException.compareAndSet(null, new StreamsException(e, task.id()));
                tasksToCloseDirty.add(task);
            }
        }

        // If any active tasks can't be committed, none of them can be, and all that need a commit must be closed dirty
        if (processingMode == EXACTLY_ONCE_V2 && !tasksToCloseDirty.isEmpty()) {
            tasksToCloseClean.removeAll(tasksToCommit);
            tasksToCloseDirty.addAll(tasksToCommit);
        } else {
            try {
                taskExecutor.commitOffsetsOrTransaction(consumedOffsetsAndMetadataPerTask);
            } catch (final RuntimeException e) {
                log.error("Exception caught while committing tasks " + consumedOffsetsAndMetadataPerTask.keySet(), e);
                // TODO: should record the task ids when handling this exception
                maybeSetFirstException(false, e, firstException);

                if (e instanceof TaskCorruptedException) {
                    final TaskCorruptedException taskCorruptedException = (TaskCorruptedException) e;
                    final Set<TaskId> corruptedTaskIds = taskCorruptedException.corruptedTasks();
                    final Set<Task> corruptedTasks = tasksToCommit
                        .stream()
                        .filter(task -> corruptedTaskIds.contains(task.id()))
                        .collect(Collectors.toSet());
                    tasksToCloseClean.removeAll(corruptedTasks);
                    tasksToCloseDirty.addAll(corruptedTasks);
                } else {
                    // If the commit fails, everyone who participated in it must be closed dirty
                    tasksToCloseClean.removeAll(tasksToCommit);
                    tasksToCloseDirty.addAll(tasksToCommit);
                }
            }

            for (final Task task : activeTaskIterable()) {
                try {
                    task.postCommit(true);
                } catch (final RuntimeException e) {
                    log.error("Exception caught while post-committing task " + task.id(), e);
                    maybeSetFirstException(false, maybeWrapTaskException(e, task.id()), firstException);
                    tasksToCloseDirty.add(task);
                    tasksToCloseClean.remove(task);
                }
            }
        }

        for (final Task task : tasksToCloseClean) {
            try {
                task.suspend();
                closeTaskClean(task);
            } catch (final RuntimeException e) {
                log.error("Exception caught while clean-closing active task {}: {}", task.id(), e.getMessage());

                if (task.state() != State.CLOSED) {
                    tasksToCloseDirty.add(task);
                }
                // ignore task migrated exception as it doesn't matter during shutdown
                maybeSetFirstException(true, maybeWrapTaskException(e, task.id()), firstException);
            }
        }

        return tasksToCloseDirty;
    }
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