shashankhs11 commented on code in PR #21059:
URL: https://github.com/apache/kafka/pull/21059#discussion_r2582658594
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1793,24 +1713,18 @@ Set<TaskId> standbyTaskIds() {
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.tasks().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();
- }
+ final Map<TaskId, Task> ret =
stateUpdater.tasks().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;
}
/**
* 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.
*
- * TODO: after we complete switching to state updater, we could rename
this function as allRunningTasks
- * to be differentiated from allTasks including running and
restoring tasks
*/
- Map<TaskId, Task> allOwnedTasks() {
+ Map<TaskId, Task> allRunningTasks() {
Review Comment:
Completed `TODO:` Replaced all occurences of `allOwnedTasks` with
`allRunningTasks`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
