private void addToTasksToClose(final Map<TaskId, CompletableFuture<StateUpdater.RemovedTaskResult>> futures,
                                   final Set<Task> tasksToCloseCleanFromStateUpdater,
                                   final Set<Task> tasksToCloseDirtyFromStateUpdater) {
        futures.entrySet().stream()
            .map(entry -> waitForFuture(entry.getKey(), entry.getValue()))
            .filter(Objects::nonNull)
            .forEach(removedTaskResult -> {
                if (removedTaskResult.exception().isPresent()) {
                    tasksToCloseDirtyFromStateUpdater.add(removedTaskResult.task());
                } else {
                    tasksToCloseCleanFromStateUpdater.add(removedTaskResult.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