{
  "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": "Deprecated: this field is not used after 4.0. The number of partitions of the topic." }, <-- deprecated field
      { "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." }
      ]}, <-- deprecated field
      { "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." } <-- new field
    ]}
  ]
}

{
  "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": "Deprecated: this field is not used after 4.0. The number of partitions of the topic." }, <-- deprecated field
      { "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." }
      ]}, <-- deprecated field
      { "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." } <-- new field
    ]}
  ]
}

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;

        // Initialize the last offset if it was not yet.
        if (lastMetadataImageWithNewTopics == -1L) {
            lastMetadataImageWithNewTopics = metadataImage.provenance().lastContainedOffset();
        }

        TopicsDelta topicsDelta = delta.topicsDelta();
        if (topicsDelta == null) return;

        // Updated the last offset of the image with newly created topics. This is used to
        // trigger a refresh of all the regular expressions when topics are created. Note
        // that we don't trigger a refresh when topics are deleted. Those are removed from
        // 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();
        }

        // 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();
            // trigger recalculate topic hash in next consumer group heartbeat
            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 {
    // ...

    /**
     * Computes the subscription metadata based on the current subscription info.
     * 
     * @param subscribedTopicNames Map of topic names to the number of subscribers. 
     * @param metadataImage        The current metadata for all available topics.
     * @param topicHashCache       The cache of topic hashes.
     * @return An immutable map of subscription topic hash for each topic that the consumer group is subscribed to.
     */
    public Map<String, TopicMetadata> computeSubscriptionMetadata(
        Map<String, SubscriptionCount> subscribedTopicNames,
        MetadataImage metadataImage,
        Map<String, Long> topicHashCache
    ) {
        // Create the topic metadata for each subscribed topic.
        Map<String, TopicMetadata> newSubscriptionMetadata = new HashMap<>(subscribedTopicNames.size());
        TopicsImage topicsImage = metadataImage.topics();
        subscribedTopicNames.forEach((topicName, count) -> {
            TopicImage topicImage = topicsImage.getTopic(topicName);
            if (topicImage != null) {
                newSubscriptionMetadata.put(topicName, new TopicMetadata(
                    topicImage.id(),
                    topicImage.name(),
                    topicHashCache.computeIfAbsent(
                        topicName,
                        key -> computeTopicHash(topicImage, metadataImage.cluster())
                    )
                ));
            }
        });

        return Collections.unmodifiableMap(newSubscriptionMetadata);
    }

    /**
     * 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.
     */
    public static 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;
    }
}