apoorvmittal10 commented on code in PR #22714:
URL: https://github.com/apache/kafka/pull/22714#discussion_r3505539222
##########
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:
nit: Just to avoid unneccessary HashSet.
```
Set<Uuid> alreadyKnownTopics = null;
if (shareGroupStatePartitionMetadata.containsKey(groupId)) {
ShareGroupStatePartitionMetadataInfo info =
shareGroupStatePartitionMetadata.get(groupId);
alreadyKnownTopics = new HashSet<>();
alreadyKnownTopics.addAll(info.initializedTopics().keySet());
alreadyKnownTopics.addAll(info.initializingTopics().keySet());
}
##########
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());
+ }
+
+ return Optional.of(buildInitializeShareGroupStateRequest(groupId,
groupEpoch, topicPartitionChangeMap, alreadyKnownTopics));
Review Comment:
nit: and here if above change looks good
```suggestion
return Optional.of(buildInitializeShareGroupStateRequest(groupId,
groupEpoch, topicPartitionChangeMap, alreadyKnownTopics != null ?
alreadyKnownTopics : Set.of()));
```
--
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]
