chia7712 commented on code in PR #17444:
URL: https://github.com/apache/kafka/pull/17444#discussion_r1821122058
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Utils.java:
##########
@@ -235,4 +238,20 @@ public static ApiMessage
messageOrNull(ApiMessageAndVersion apiMessageAndVersion
return apiMessageAndVersion.message();
}
}
+
+ public static long hashSubscriptionMetadata(Map<String, TopicMetadata>
subscriptionMetadata) {
Review Comment:
> For the coordinator, it's in memory cache. When a coordinator starts, we
will recalculate it from metadata image in
GroupMetadataManager#subscribeGroupToTopic
This might be too detailed, but perhaps we don't need to build the hash
cache at this stage. Instead, we could build the cache in
`GroupMetadataManager#consumerGroupHeartbeat` when it's actually needed. For
example:
```java
public Map<String, Integer> computeSubscriptionMetadata(
Map<String, Integer> subscribedTopicNames,
TopicsImage topicsImage,
Map<String, Integer> cache
) {
// Create the topic metadata for each subscribed topic.
Map<String, Integer> newSubscriptionMetadata = new
HashMap<>(subscribedTopicNames.size());
subscribedTopicNames.forEach((topicName, count) -> {
TopicImage topicImage = topicsImage.getTopic(topicName);
if (topicImage != null) {
var topicMetadata = new TopicMetadata(
topicImage.id(),
topicImage.name(),
topicImage.partitions().size()
);
newSubscriptionMetadata.put(topicName,
cache.computeIfAbsent(topicName, __ -> topicMetadata.hashCode()));
}
});
return Collections.unmodifiableMap(newSubscriptionMetadata);
}
```
--
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]
