sjhajharia commented on code in PR #22714:
URL: https://github.com/apache/kafka/pull/22714#discussion_r3506437792
##########
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);
+ Set<Uuid> alreadyKnownTopics = new HashSet<>();
+ if (info != null) {
+ alreadyKnownTopics.addAll(info.initializedTopics().keySet());
+ alreadyKnownTopics.addAll(info.initializingTopics().keySet());
+ }
Review Comment:
Thanks for the review. I have updated the PR accordingly.
--
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]
