zzbennett commented on code in PR #20061:
URL: https://github.com/apache/kafka/pull/20061#discussion_r2182089540
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -5725,24 +5717,23 @@ public void onNewMetadataImage(MetadataImage newImage,
MetadataDelta delta) {
// the subscription metadata (and the assignment) via the above
mechanism. The
// resolved regular expressions are cleaned up on the next refresh.
if (!topicsDelta.createdTopicIds().isEmpty()) {
- lastMetadataImageWithNewTopics =
metadataImage.provenance().lastContainedOffset();
+ lastMetadataImageWithNewTopics = metadataImage.version();
}
// Notify all the groups subscribed to the created, updated or
// deleted topics.
Set<String> allGroupIds = new HashSet<>();
- topicsDelta.changedTopics().forEach((topicId, topicDelta) -> {
- String topicName = topicDelta.name();
- // Remove topic hash from the cache to recalculate it.
- topicHashCache.remove(topicName);
- allGroupIds.addAll(groupsSubscribedToTopic(topicName));
- });
- topicsDelta.deletedTopicIds().forEach(topicId -> {
- TopicImage topicImage = delta.image().topics().getTopic(topicId);
- String topicName = topicImage.name();
- topicHashCache.remove(topicName);
- allGroupIds.addAll(groupsSubscribedToTopic(topicName));
- });
+ topicsDelta.changedTopicIds().forEach(topicId ->
+ metadataImage.topicName(topicId).ifPresent(topicName -> {
+ // Remove topic hash from the cache to recalculate it.
+ topicHashCache.remove(topicName);
+ allGroupIds.addAll(groupsSubscribedToTopic(topicName));
+ }));
+ topicsDelta.deletedTopicIds().forEach(topicId ->
+ delta.image().topicName(topicId).ifPresent(topicName -> {
Review Comment:
Added a javadoc to clarify that the image returned from `delta.image()`
needs be a snapshot of the metadata _before_ the delta occurred
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
