aliehsaeedii commented on code in PR #22808:
URL: https://github.com/apache/kafka/pull/22808#discussion_r3585454670
##########
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:
Reading at latest instead of `committedOffset` can surface offsets from a
heartbeat whose write later failed and rolled back — `taskOffsets` isn't a
timeline map, so it's never reverted. Low impact since it's transient and
overwritten on the next heartbeat, but the describe output can briefly show an
uncommitted value. Maybe we can ignore this (no need to fix) but document it
some where?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMember.java:
##########
@@ -407,6 +417,16 @@ public StreamsGroupDescribeResponseData.Member
asStreamsGroupDescribeMember(Task
);
}
+ private static List<StreamsGroupDescribeResponseData.TaskOffset>
taskOffsetsFromMap(Map<TaskId, Long> offsets) {
+ return offsets.entrySet().stream()
+ .sorted(Map.Entry.comparingByKey())
Review Comment:
There’s no test specifically exercising the `sorted()` behavior. Please add
a case with out-of-order input to lock the ordering in.
--
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]
