Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java (Java)

Code before:
                }
                // ignore task migrated exception as it doesn't matter during shutdown
                maybeSetFirstException(true, maybeWrapTaskException(e, task.id()), firstException);
            }
        }
        return tasksToCloseDirty;
    }

    Set<TaskId> activeTaskIds() {
        return activeTaskStream()
            .map(Task::id)
            .collect(Collectors.toSet());
    }

    Set<TaskId> standbyTaskIds() {
        return standbyTaskStream()
            .map(Task::id)
            .collect(Collectors.toSet());
    }

    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());
            return ret;
        } else {
            return tasks.allTasksPerId();
        }
    }

    Map<TaskId, Task> notPausedTasks() {
        return Collections.unmodifiableMap(tasks.allTasks()
            .stream()
            .filter(t -> !topologyMetadata.isPaused(t.id().topologyName()))
            .collect(Collectors.toMap(Task::id, v -> v)));
    }

    Map<TaskId, Task> activeTaskMap() {
        return activeTaskStream().collect(Collectors.toMap(Task::id, t -> t));
    }

    List<Task> activeTaskIterable() {
        return activeTaskStream().collect(Collectors.toList());
    }

    private Stream<Task> activeTaskStream() {
        return tasks.allTasks().stream().filter(Task::isActive);
    }


Review comment (suggestion):
I've changed the func name slightly in another PR, so if that PR is merged we need to do a slight rebase/conflict resolution, just FYI.

Code after:
                }
                // ignore task migrated exception as it doesn't matter during shutdown
                maybeSetFirstException(true, maybeWrapTaskException(e, task.id()), firstException);
            }
        }
        return tasksToCloseDirty;
    }

    Set<TaskId> activeTaskIds() {
        return activeTaskStream()
            .map(Task::id)
            .collect(Collectors.toSet());
    }

    Set<TaskId> standbyTaskIds() {
        return standbyTaskStream()
            .map(Task::id)
            .collect(Collectors.toSet());
    }

    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());
            return ret;
        } else {
            return tasks.allTasksPerId();
        }
    }

    /**
     * Returns tasks owned by the stream thread. With state updater disabled, these are all tasks. With
     * state updater enabled, this does not return any tasks currently owned by the state updater.
     * @return
     */
    Map<TaskId, Task> allOwnedTasks() {
        // 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.
        return tasks.allTasksPerId();
    }

    Set<Task> readOnlyAllTasks() {
        // 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 HashSet<Task> ret = new HashSet<>(stateUpdater.getTasks());
            ret.addAll(tasks.allTasks());
            return Collections.unmodifiableSet(ret);
        } else {