Map<TaskId, Task> allTasks() {
        // not bothering with an unmodifiable map, since the tasks themselves are mutable, but
        // if any outside code modifies the map or the tasks, it would be a severe transgression.
        if (stateUpdater != null) {
            final Map<TaskId, Task> ret = stateUpdater.getTasks().stream().collect(Collectors.toMap(Task::id, x -> x));
            ret.putAll(tasks.allTasksPerId());
            ret.putAll(tasks.pendingTasksToInit().stream().collect(Collectors.toMap(Task::id, x -> x)));
            return ret;
        } else {
            return tasks.allTasksPerId();
        }
    }
Closes and cleans up after the provided tasks, including closing their corresponding task producers
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java