mjsax commented on code in PR #23082:
URL: https://github.com/apache/kafka/pull/23082#discussion_r3723380368
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1323,11 +1327,20 @@ public Map<TaskId, Long> taskOffsetSums() {
final Map<TaskId, Long> taskOffsetSums =
stateDirectory.taskOffsetSums(lockedTaskDirectoriesOfNonOwnedTasksAndClosedAndCreatedTasks);
- // overlay latest offsets from assigned tasks
+ // Overlay latest offsets from assigned tasks.
+ // `stateDirectory.taskOffsetSums` above only cover what is
recoverable from disk (potentially stale),
+ // including offsets from persistent stores which tasks are owned by
sibling thread, as well as dormant tasks;
+ // We update this (potentially state) information with latest
changelog offset inforamtion for all other
+ // tasks assigned to this thread; this step also adds offset-sum
information for in-memory state stores
for (final Task task : tasks.values()) {
// exclude stateless and non-logged tasks
if (task.isActive() && task.state() == State.RUNNING &&
!task.changelogPartitions().isEmpty()) {
taskOffsetSums.put(task.id(), Task.LATEST_OFFSET);
Review Comment:
The main point of the fix is actually _dormant_ tasks, ie, not owned any
longer. For persistent stores, we still have the state in the state directory,
so if the state-directory caches them that's correct. However, for in-memory
stores, when a task is revoked we don't have any local state left, but the
state-store cache still reported these offsets.
For assigned in-memory tasks (active or standby) it's ok if we report
offsets, and not much changes for them with this PR -- before the change, the
in-memory offsets are pushed into the cache and read from the cache for
reporting. With the change, we won't add these offset into the cache any
longer, and thus need to add them back on the read path for reporting.
The point is, that we don't report offset of dormet/previously assigned
tasks with in-memory offsets any longer -- they don't make it into the cache
any more, and on-read we only go over assigned tasks.
--
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]
