mjsax commented on code in PR #22808:
URL: https://github.com/apache/kafka/pull/22808#discussion_r3617387222
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java:
##########
@@ -222,6 +222,14 @@ public static class DeadlineAndEpoch {
* This is transient telemetry that the assignor uses to estimate task lag
for warm-up promotion. Per KIP-1071
* it is not persisted to the {@code __consumer_offsets} topic; it is held
in memory and re-reported by members
* on the task-offset interval (and is therefore lost on coordinator
failover until re-reported).
+ *
+ * <p>Unlike the rest of the member state this is a plain (non-timeline)
map, so it does not take part in the
+ * coordinator's snapshot/rollback machinery. A value written while
handling a heartbeat is therefore <em>not</em>
+ * reverted if that heartbeat's record append later fails and the timeline
state rolls back to
+ * {@code committedOffset}; it lingers until the member's next heartbeat
overwrites it. Since
+ * {@link #asDescribedGroup(long)} reads these at the latest in-memory
value, a describe issued in that window can
+ * momentarily surface a task offset that was never committed. This is
intentional: the data is transient and
+ * self-heals on the next heartbeat, so it is not worth the cost of making
it timeline-aware.
Review Comment:
It seems this comment is overly verbose, and repeats stuff already said in
the first paragraph.
I also don't see any issue with "describe group" so not sure if we need to
add anything at all here? We could maybe extend the existing paragraph and
explain the "no rollback" behavior, in addition to "is transient" property. But
a single sentence should be sufficient. Not a whole paragraph.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java:
##########
@@ -1333,7 +1341,13 @@ 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. As a result, if a
+ // heartbeat updated them but its record append then
failed and rolled back, this can momentarily
+ // surface an offset that was never committed; it
self-heals on the next heartbeat (see the
+ // taskOffsets field for details). Members that have not
reported any yield MemberTaskOffsets.EMPTY.
Review Comment:
Similar to above. I think we can largely (on entirely?) remove this comment?
--
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]
