private void handleTasksInStateUpdater(final Map<TaskId, Set<TopicPartition>> activeTasksToCreate,
                                           final Map<TaskId, Set<TopicPartition>> standbyTasksToCreate,
                                           final Map<Task, Set<TopicPartition>> tasksToRecycle,
                                           final Set<Task> tasksToCloseCleanFromStateUpdater,
                                           final Set<Task> tasksToCloseDirtyFromStateUpdater,
                                           final Map<TaskId, RuntimeException> failedTasks) {
        final Map<TaskId, Set<TopicPartition>> newInputPartitions = new HashMap<>();
        final Map<TaskId, Set<TopicPartition>> standbyInputPartitions = new HashMap<>();
        final Map<TaskId, Set<TopicPartition>> activeInputPartitions = new HashMap<>();
        final Map<TaskId, CompletableFuture<StateUpdater.RemovedTaskResult>> futuresForUpdatingInputPartitions = new LinkedHashMap<>();
        final Map<TaskId, CompletableFuture<StateUpdater.RemovedTaskResult>> futuresForActiveTasksToRecycle = new LinkedHashMap<>();
        final Map<TaskId, CompletableFuture<StateUpdater.RemovedTaskResult>> futuresForStandbyTasksToRecycle = new LinkedHashMap<>();
        final Map<TaskId, CompletableFuture<StateUpdater.RemovedTaskResult>> futuresForTasksToClose = new LinkedHashMap<>();
        for (final Task task : stateUpdater.getTasks()) {
            final TaskId taskId = task.id();
            if (activeTasksToCreate.containsKey(taskId)) {
                if (task.isActive()) {
                    if (!task.inputPartitions().equals(activeTasksToCreate.get(taskId))) {
                        final CompletableFuture<StateUpdater.RemovedTaskResult> future = stateUpdater.remove(taskId);
                        futuresForUpdatingInputPartitions.put(taskId, future);
                        newInputPartitions.put(taskId, activeTasksToCreate.get(taskId));
                    }
                } else {
                    final CompletableFuture<StateUpdater.RemovedTaskResult> future = stateUpdater.remove(taskId);
                    futuresForStandbyTasksToRecycle.put(taskId, future);
                    activeInputPartitions.put(taskId, activeTasksToCreate.get(taskId));
                }
                activeTasksToCreate.remove(taskId);
            } else if (standbyTasksToCreate.containsKey(taskId)) {
                if (task.isActive()) {
                    final CompletableFuture<StateUpdater.RemovedTaskResult> future = stateUpdater.remove(taskId);
                    futuresForActiveTasksToRecycle.put(taskId, future);
                    standbyInputPartitions.put(taskId, standbyTasksToCreate.get(taskId));
                }
                standbyTasksToCreate.remove(taskId);
            } else {
                final CompletableFuture<StateUpdater.RemovedTaskResult> future = stateUpdater.remove(taskId);
                futuresForTasksToClose.put(taskId, future);
            }
        }
        updateInputPartitions(futuresForUpdatingInputPartitions, newInputPartitions, failedTasks);
        addToActiveTasksToRecycle(futuresForActiveTasksToRecycle, standbyInputPartitions, tasksToRecycle, failedTasks);
        addToStandbyTasksToRecycle(futuresForStandbyTasksToRecycle, activeInputPartitions, tasksToRecycle, failedTasks);
        addToTasksToClose(futuresForTasksToClose, tasksToCloseCleanFromStateUpdater, tasksToCloseDirtyFromStateUpdater);
    }
@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