mjsax commented on code in PR #22644:
URL: https://github.com/apache/kafka/pull/22644#discussion_r3468642999
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2118,6 +2121,14 @@ private CoordinatorResult<StreamsGroupHeartbeatResult,
CoordinatorRecord> stream
);
}
+ // Store the latest task changelog offsets/end-offsets reported by the
member. These are transient telemetry
+ // (used by the assignor to estimate task lag) and are not persisted.
Task offsets and end-offsets are reported
+ // independently: a null list means "unchanged since the last
heartbeat", so we retain the previously reported
+ // value for whichever of the two is null and only update when at
least one is reported.
+ if (taskOffsets != null || taskEndOffsets != null) {
+ group.updateTaskOffsets(memberId,
group.taskOffsets(memberId).update(taskOffsets, taskEndOffsets));
Review Comment:
Ah. Good catch.
--
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]
