smjn commented on code in PR #22714:
URL: https://github.com/apache/kafka/pull/22714#discussion_r3503970157
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3133,18 +3133,41 @@ private Optional<InitializeShareGroupStateParameters>
maybeCreateInitializeShare
}
addInitializingTopicsRecords(groupId, records,
topicPartitionChangeMap);
- return Optional.of(buildInitializeShareGroupStateRequest(groupId,
groupEpoch, topicPartitionChangeMap));
+
+ // Topics already known to the group (already initialized or already
initializing) whose
+ // newly added partitions are now being initialized must start at
offset 0 to avoid losing
+ // records produced before initialization completes. Brand-new topics
(seen for the first
+ // time, present in neither map) use -1 so that the group's
share.auto.offset.reset strategy
+ // applies. This is read before the records above are replayed, so a
brand-new topic is
+ // correctly absent here.
+ ShareGroupStatePartitionMetadataInfo info =
shareGroupStatePartitionMetadata.get(groupId);
Review Comment:
This looks good since if and when a request with new partitions on existing
topic comes, it implies that previous requests have already produced the group
coord records which have been updated in the info map.
--
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]
