private void handleRunningAndSuspendedTasks(final Map<TaskId, Set<TopicPartition>> activeTasksToCreate,
                                                final Map<TaskId, Set<TopicPartition>> standbyTasksToCreate,
                                                final Map<Task, Set<TopicPartition>> tasksToRecycle,
                                                final Set<Task> tasksToCloseClean) {
        for (final Task task : tasks.allTasks()) {
            if (!task.isActive()) {
                throw new IllegalStateException("Standby tasks should only be managed by the state updater, " +
                    "but standby task " + task.id() + " is managed by the stream thread");
            }
            final TaskId taskId = task.id();
            if (activeTasksToCreate.containsKey(taskId)) {
                handleReassignedActiveTask(task, activeTasksToCreate.get(taskId));
                activeTasksToCreate.remove(taskId);
            } else if (standbyTasksToCreate.containsKey(taskId)) {
                tasksToRecycle.put(task, standbyTasksToCreate.get(taskId));
                standbyTasksToCreate.remove(taskId);
            } else {
                tasksToCloseClean.add(task);
            }
        }
    }
@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