private Stream<Task> getNonFailedTasks(final Map<TaskId, CompletableFuture<StateUpdater.RemovedTaskResult>> futures,
                                           final Map<TaskId, RuntimeException> failedTasks) {
        return futures.entrySet().stream()
            .map(entry -> waitForFuture(entry.getKey(), entry.getValue()))
            .filter(Objects::nonNull)
            .map(removedTaskResult -> checkIfTaskFailed(removedTaskResult, failedTasks))
            .filter(Objects::nonNull);
    }
@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