private void removeLostActiveTasksFromStateUpdaterAndPendingTasksToInit() {
        if (stateUpdater != null) {
            final Map<TaskId, CompletableFuture<StateUpdater.RemovedTaskResult>> futures = new LinkedHashMap<>();
            final Map<TaskId, RuntimeException> failedTasksDuringCleanClose = new HashMap<>();
            final Set<Task> tasksToCloseClean = new HashSet<>(tasks.drainPendingActiveTasksToInit());
            final Set<Task> tasksToCloseDirty = new HashSet<>();
            for (final Task restoringTask : stateUpdater.getTasks()) {
                if (restoringTask.isActive()) {
                    futures.put(restoringTask.id(), stateUpdater.remove(restoringTask.id()));
                }
            }

            addToTasksToClose(futures, tasksToCloseClean, tasksToCloseDirty);
            for (final Task task : tasksToCloseClean) {
                closeTaskClean(task, tasksToCloseDirty, failedTasksDuringCleanClose);
            }
            for (final Task task : tasksToCloseDirty) {
                closeTaskDirty(task, false);
            }
        }
    }
Closes active tasks as zombies, as these partitions have been lost and are no longer owned.
NOTE this method assumes that when it is called, EVERY task/partition has been lost and must
be closed as a zombie.

@throws TaskMigratedException if the task producer got fenced (EOS only)
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java