aliehsaeedii commented on code in PR #22808:
URL: https://github.com/apache/kafka/pull/22808#discussion_r3616180191
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java:
##########
@@ -1308,7 +1308,11 @@ public StreamsGroupDescribeResponseData.DescribedGroup
asDescribedGroup(
members.entrySet(committedOffset).forEach(
entry -> describedGroup.members().add(
entry.getValue().asStreamsGroupDescribeMember(
- targetAssignment.get(entry.getValue().memberId(),
committedOffset)
+ targetAssignment.get(entry.getValue().memberId(),
committedOffset),
+ // Task (end-)offsets are transient, unpersisted
telemetry, so unlike the rest of the member state
+ // they are read from the latest in-memory value rather
than at committedOffset. Members that have
+ // not reported any yield MemberTaskOffsets.EMPTY.
+ taskOffsets(entry.getValue().memberId())
Review Comment:
By "offset" I mean the per-task changelog offset/end-offset the member
reports in its heartbeat — the MemberTaskOffsets (taskOffsets/taskEndOffsets)
that the assignor uses to estimate task lag.
Concrete example. Member M owns task 0_0:
1. Last committed heartbeat reported offset 100 for 0_0.
2. A new heartbeat arrives reporting 0_0 = 150. The handler calls
group.updateTaskOffsets(...), so the in-memory map is now 150. The same
heartbeat also produces coordinator records (epoch bump / assignment change /
etc.) to be persisted.
3. The append to __consumer_offsets fails (for whatever reason). The runtime
rolls back timeline state to committedOffset — member epoch, assignment,
everything reverts to the pre-heartbeat snapshot (the 100-world). But
taskOffsets is a plain HashMap, not a TimelineObject, so it is not rolled back
— it keeps 150.
4. A StreamsGroupDescribe now comes in. Everything else on the member is
read at committedOffset (the consistent, reverted snapshot), but
taskOffsets(memberId) reads the plain map → 150.
Net effect: describe momentarily reports a task changelog offset (and hence
a lag) that was never actually committed. It's really transient and self-heals
on the next successful heartbeat, which overwrites the map — so I don't think
it's worth a fix, I just wanted to raise it.
--
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]
