{
  "type": "data",
  "name": "ConsumerGroupPartitionMetadataValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Topics", "versions": "0+", "type": "[]TopicMetadata",
      "about": "The list of topic metadata.", "fields": [
      { "name": "TopicId", "versions": "0+", "type": "uuid",
        "about": "The topic id." },
      { "name": "TopicName", "versions": "0+", "type": "string",
        "about": "The topic name." },
      { "name": "NumPartitions", "versions": "0+", "type": "int32",
        "about": "The number of partitions of the topic." },
      { "name": "PartitionMetadata", "versions": "0+", "type": "[]PartitionMetadata",
        "about": "Deprecated: this field is not used after 4.0. Partitions mapped to a set of racks. If the rack information is unavailable for all the partitions, an empty list is stored", "fields": [
          { "name": "Partition", "versions": "0+", "type": "int32",
            "about": "The partition number." },
          { "name": "Racks", "versions": "0+", "type": "[]string",
            "about": "The set of racks that the partition is mapped to." }
      ]},
      { "name": "TopicHash", "versions": "0+", "type": "int64",
        "default": 0, "taggedVersions": "0+", "tag": 0,
        "about": "The hash value of the topic id, name, number of partitions, and partition racks." }
    ]}
  ]
}

{
  "type": "data",
  "name": "ShareGroupPartitionMetadataValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Topics", "versions": "0+", "type": "[]TopicMetadata",
      "about": "The list of topic metadata.", "fields": [
      { "name": "TopicId", "versions": "0+", "type": "uuid",
        "about": "The topic id." },
      { "name": "TopicName", "versions": "0+", "type": "string",
        "about": "The topic name." },
      { "name": "NumPartitions", "versions": "0+", "type": "int32",
        "about": "The number of partitions of the topic." },
      { "name": "PartitionMetadata", "versions": "0+", "type": "[]PartitionMetadata",
        "about": "Partitions mapped to a set of racks. If the rack information is unavailable for all the partitions, an empty list is stored", "fields": [
        { "name": "Partition", "versions": "0+", "type": "int32",
          "about": "The partition number." },
        { "name": "Racks", "versions": "0+", "type": "[]string",
          "about": "The set of racks that the partition is mapped to." }
      ]},
      { "name": "TopicHash", "versions": "0+", "type": "int64",
        "default": 0, "taggedVersions": "0+", "tag": 0,
        "about": "The hash value of the topic id, name, number of partitions, and partition racks." }
    ]}
  ]
}

public class GroupMetadataManager {
    // ...

    /**
     * The topic hash value by topic name.
     */
    private final Map<String, Long> topicHashCache; 

    /**
     * Unsubscribes a group from a topic.
     *
     * @param groupId   The group id.
     * @param topicName The topic name.
     */
    private void unsubscribeGroupFromTopic(
        String groupId,
        String topicName
    ) {
        groupsByTopics.computeIfPresent(topicName, (__, groupIds) -> {
            groupIds.remove(groupId);
            if (groups.isEmpty()) {
                topicHashCache.remove(topicName);
                return null;
            }
            return groupIds;
        });
    }

    /**
     * A new metadata image is available.
     *
     * @param newImage  The new metadata image.
     * @param delta     The delta image.
     */
    public void onNewMetadataImage(MetadataImage newImage, MetadataDelta delta) {
        metadataImage = newImage;

        // Notify all the groups subscribed to the created, updated or
        // deleted topics.
        Optional.ofNullable(delta.topicsDelta()).ifPresent(topicsDelta -> {
            Set<String> allGroupIds = new HashSet<>();
            topicsDelta.changedTopics().forEach((topicId, topicDelta) -> {
                String topicName = topicDelta.name();
                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));
            });
            allGroupIds.forEach(groupId -> {
                Group group = groups.get(groupId);
                if (group != null && (group.type() == CONSUMER || group.type() == SHARE)) {
                    ((ModernGroup<?>) group).requestMetadataRefresh();
                }
            });
        });
    }
}

public abstract class ModernGroup<T extends ModernGroupMember> implements Group {
    // ...

    /**
     * The hash with each subscribed topic name.
     */
    protected final TimelineHashMap<String, Long> subscriptionTopicHash;

    /**
     * @return An immutable Map of subscription topic hash for
     *         each topic that the consumer group is subscribed to.
     */
    public Map<String, Long> subscriptionTopicHash() {
        return Collections.unmodifiableMap(subscriptionTopicHash);
    }


    /**
     * Updates the subscription topic hash. This replaces the previous one.
     *
     * @param subscriptionTopicHash The new subscription topic hash.
     */
    public void setSubscriptionTopicHash(
        Map<String, Long> subscriptionTopicHash
    ) {
        this.subscriptionTopicHash.clear();
        this.subscriptionTopicHash.putAll(subscriptionTopicHash);
    }

    /**
     * Computes the subscription topic hash based on the current subscription info.
     * 
     * @param subscribedTopicNames  Map of topic names to the number of subscribers.
     * @param topicHashCache        The current topic hash cache from coordinator.
     *
     * @return An immutable map of subscription topic hash for each topic that the consumer group is subscribed to.
     */
    public Map<String, Long> computeSubscriptionTopicHash(
        Map<String, Integer> subscribedTopicNames,
        TopicsImage topicsImage,
        ClusterImage clusterImage,
        Map<String, Long> topicHashCache
    ) {
        // Create the topic hash for each subscribed topic.
        Map<String, Long> newSubscriptionTopicHash = new HashMap<>();
        subscribedTopicNames.forEach((topicName, count) -> {
            TopicImage topicImage = topicsImage.getTopic(topicName);
            if (topicImage != null) {
                newSubscriptionTopicHash.put(topicName, cache.computeIfAbsent(topicName, t -> computeTopicHash(topicImage, clusterImage)));
            }
        });
        return Collections.unmodifiableMap(newSubscriptionTopicHash);
    }

    /**
     * Computes the hash of the topic id, name, number of partitions, and partition racks by Murmur3.
     *
     * @param topicImage   The topic image.
     * @param clusterImage The cluster image.
     */
    private long computeTopicHash(TopicImage topicImage, ClusterImage clusterImage) {
        long result = topicImage.id().hashCode();
        result = 63 * result + Murmur3.hash64(topicImage.name().getBytes(StandardCharsets.UTF_8));
        result = 63 * result + Murmur3.hash64(topicImage.partitions().size());
        result = 63 * result + topicImage.partitions().entrySet().stream().mapToLong(
            entry -> {
                PartitionRegistration partitionRegistration = entry.getValue();
                long partitionHash = Murmur3.hash64(entry.getKey());
                Set<String> racks = Arrays.stream(partitionRegistration.replicas)
                    .mapToObj(clusterImage::broker)
                    .map(BrokerRegistration::rack)
                    .filter(Optional::isPresent)
                    .map(Optional::get)
                    .collect(Collectors.toSet());
                return 63 * partitionHash + racks.stream().mapToLong(
                    rack -> Murmur3.hash64(rack.getBytes(StandardCharsets.UTF_8))
                ).reduce(0L, Long::sum);
            }
        ).reduce(0L, Long::sum);
        return result;
    } }