[GitHub] [kafka] cadonna commented on a diff in pull request #13925: KAFKA-10199: Consider tasks in state updater when computing offset sums

["cadonna (via GitHub)" <gi...@apache.org>]

cadonna commented on code in PR #13925:
URL: https://github.com/apache/kafka/pull/13925#discussion_r1245477150


##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1138,28 +1138,33 @@ public void signalResume() {
     public Map<TaskId, Long> getTaskOffsetSums() {
         final Map<TaskId, Long> taskOffsetSums = new HashMap<>();
 
-        // Not all tasks will create directories, and there may be directories for tasks we don't currently own,
-        // so we consider all tasks that are either owned or on disk. This includes stateless tasks, which should
-        // just have an empty changelogOffsets map.
-        for (final TaskId id : union(HashSet::new, lockedTaskDirectories, tasks.allTaskIds())) {
-            final Task task = tasks.contains(id) ? tasks.task(id) : null;
-            // Closed and uninitialized tasks don't have any offsets so we should read directly from the checkpoint
-            if (task != null && task.state() != State.CREATED && task.state() != State.CLOSED) {
+        final Map<TaskId, Task> tasks = allTasks();
+        final Set<TaskId> lockedTaskDirectoriesOfNonOwnedTasksAndClosedAndCreatedTasks =

Review Comment:
   I do not think there is guarantee that `lockedTaskDirectories` contains any tasks the client owns. `lockedTaskDirectories` are just the non-empty task directories in the state directory when a rebalance starts. However, a task directory is created when a task is created, i.e., it is in state `CREATE`. A task directory is not deleted when a task is closed, i.e., in state `CLOSED`. This might be a correlation and not a thought-out invariant. At least, the original code did not rely on this since it used `union(HashSet::new, lockedTaskDirectories, tasks.allTaskIds())`.
   I am also somehow reluctant to rely on such -- IMO -- brittle invariant. 
   The creation of the task directory can probably be moved to other parts of the code like when the task is initialized which would mean that there is a interval in which the task is in state `CREATED` but does not have a task directory.



-- 
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: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org