private Map<TaskId, RuntimeException> closeAndRecycleTasks(final Map<Task, Set<TopicPartition>> tasksToRecycle,
                                                               final Set<Task> tasksToCloseClean) {
        final Map<TaskId, RuntimeException> taskCloseExceptions = new LinkedHashMap<>();
        final Set<Task> tasksToCloseDirty = new TreeSet<>(Comparator.comparing(Task::id));

        // for all tasks to close or recycle, we should first write a checkpoint as in post-commit
        final List<Task> tasksToCheckpoint = new ArrayList<>(tasksToCloseClean);
        tasksToCheckpoint.addAll(tasksToRecycle.keySet());
        for (final Task task : tasksToCheckpoint) {
            try {
                // Note that we are not actually committing here but just check if we need to write checkpoint file:
                // 1) for active tasks prepareCommit should return empty if it has committed during suspension successfully,
                //    and their changelog positions should not change at all postCommit would not write the checkpoint again.
                // 2) for standby tasks prepareCommit should always return empty, and then in postCommit we would probably
                //    write the checkpoint file.
                final Map<TopicPartition, OffsetAndMetadata> offsets = task.prepareCommit();
                if (!offsets.isEmpty()) {
                    log.error("Task {} should have been committed when it was suspended, but it reports non-empty " +
                                    "offsets {} to commit; this means it failed during last commit and hence should be closed dirty",
                            task.id(), offsets);

                    tasksToCloseDirty.add(task);
                } else if (!task.isActive()) {
                    // For standby tasks, always try to first suspend before committing (checkpointing) it;
                    // Since standby tasks do not actually need to commit offsets but only need to
                    // flush / checkpoint state stores, so we only need to call postCommit here.
                    task.suspend();

                    task.postCommit(true);
                }
            } catch (final RuntimeException e) {
                final String uncleanMessage = String.format(
                        "Failed to checkpoint task %s. Attempting to close remaining tasks before re-throwing:",
                        task.id());
                log.error(uncleanMessage, e);
                taskCloseExceptions.putIfAbsent(task.id(), e);
                // We've already recorded the exception (which is the point of clean).
                // Now, we should go ahead and complete the close because a half-closed task is no good to anyone.
                tasksToCloseDirty.add(task);
            }
        }

        tasksToCloseClean.removeAll(tasksToCloseDirty);
        for (final Task task : tasksToCloseClean) {
            try {
                closeTaskClean(task);
            } catch (final RuntimeException closeTaskException) {
                final String uncleanMessage = String.format(
                    "Failed to close task %s cleanly. Attempting to close remaining tasks before re-throwing:",
                    task.id());
                log.error(uncleanMessage, closeTaskException);

                if (task.state() != State.CLOSED) {
                    tasksToCloseDirty.add(task);
                }

                taskCloseExceptions.putIfAbsent(task.id(), closeTaskException);
            }
        }

        tasksToRecycle.keySet().removeAll(tasksToCloseDirty);
        for (final Map.Entry<Task, Set<TopicPartition>> entry : tasksToRecycle.entrySet()) {
            final Task oldTask = entry.getKey();
            final Set<TopicPartition> inputPartitions = entry.getValue();
            try {
                if (oldTask.isActive()) {
                    final StandbyTask standbyTask = convertActiveToStandby((StreamTask) oldTask, inputPartitions);
                    if (stateUpdater != null) {
                        tasks.removeTask(oldTask);
                        tasks.addPendingTasksToInit(Collections.singleton(standbyTask));
                    } else {
                        tasks.replaceActiveWithStandby(standbyTask);
                    }
                } else {
                    final StreamTask activeTask = convertStandbyToActive((StandbyTask) oldTask, inputPartitions);
                    tasks.replaceStandbyWithActive(activeTask);
                }
            } catch (final RuntimeException e) {
                final String uncleanMessage = String.format("Failed to recycle task %s cleanly. " +
                    "Attempting to close remaining tasks before re-throwing:", oldTask.id());
                log.error(uncleanMessage, e);
                taskCloseExceptions.putIfAbsent(oldTask.id(), e);
                tasksToCloseDirty.add(oldTask);
            }
        }

        // for tasks that cannot be cleanly closed or recycled, close them dirty
        for (final Task task : tasksToCloseDirty) {
            closeTaskDirty(task, true);
        }

        return taskCloseExceptions;
    }
@throws TaskMigratedException if the task producer got fenced (EOS only)
@throws StreamsException fatal error while creating / initializing the task

public for upgrade testing only
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java