private void handleRestoringAndUpdatingTasks(final Map<TaskId, Set<TopicPartition>> activeTasksToCreate,
                                                 final Map<TaskId, Set<TopicPartition>> standbyTasksToCreate,
                                                 final Map<TaskId, RuntimeException> failedTasks) {
        final Map<Task, Set<TopicPartition>> tasksToRecycleFromStateUpdater = new HashMap<>();
        final Set<Task> tasksToCloseCleanFromStateUpdater = new HashSet<>();
        final Set<Task> tasksToCloseDirtyFromStateUpdater = new HashSet<>();
        handleTasksInStateUpdater(
            activeTasksToCreate,
            standbyTasksToCreate,
            tasksToRecycleFromStateUpdater,
            tasksToCloseCleanFromStateUpdater,
            tasksToCloseDirtyFromStateUpdater,
            failedTasks
        );
        tasksToRecycleFromStateUpdater.forEach((task, inputPartitions) ->
            recycleTaskFromStateUpdater(
                task,
                inputPartitions,
                tasksToCloseDirtyFromStateUpdater,
                failedTasks
            )
        );
        tasksToCloseCleanFromStateUpdater.forEach(task ->
            closeTaskClean(task, tasksToCloseDirtyFromStateUpdater, failedTasks)
        );
        tasksToCloseDirtyFromStateUpdater.forEach(task ->
            closeTaskDirty(task, false)
        );
    }
@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