...

Following table compares cpu / memory / disk usage of different strategies:

We choose If using the second way, but not the third way, because a single hash to represent all subscribed topic in a group wastes too strategy, it still uses lot of disk size to store redundant data. For example, a group subscribes a same topic list. Every time a subscribed topic change, the group needs to store a new subscription topic metadata map with a single entry changed. If using the third strategy, it wastes too much CPU resource to recalculate the hash when the metadata image is expired. If any subscribed topic change, the third way needs to aggregate all data an recalculate the hash. The second way only needs to recalculate expired topic hash one time, because it keeps the result in coordinator cache. Please check Topic Hash Map Cache in Coordinator section for more details.

Public Interfaces

ConsumerGroupPartitionMetadataValue

Add a new tagged field TopicHash. 

In this KIP, we will combine the second and third strategies. In coordinator, there is a cache to store each topic hash. When a topic change, only single topic hash will be recalculated. Different groups can get the topic hash from the cache, so the coordinator doesn't need to calculate a same topic hash multiple times. In __consumer_offsets, the coordinator sums all subscribed topic hash to a single hash per group and store to the disk, so it doesn't uses too much disk usage to represent the change.

Public Interfaces

ConsumerGroupPartitionMetadataValue / ShareGroupPartitionMetadataValue

Both ConsumerGroupPartitionMetdataValue and ShareGroupPartitionMetadataValue will be deleted, because the coordinator doesn't store topic metadata.

ConsumerGroupMetadataValue

Add a new field "AllTopicHash".

{
  "type": "data",
  "name": "ConsumerGroupPartitionMetadataValueConsumerGroupMetadataValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "TopicsEpoch", "versions": "0+", "type": "[]TopicMetadataint32",
      "about": "The list of topic metadatagroup epoch.", "fields": [
   },
    { "name": "TopicIdAllTopicHash", "versions": "0+", "type": "uuidint64",
      "default": 0, "abouttaggedVersions": "0+"The topic id." },
 , "tag": 0,
     { "nameabout": "TopicName", "versionsThe hash of all topics in the group." } <-- new field
  ]
}

ShareGroupMetadataValue

Add a new field "AllTopicHash".

{
  "type": "0+data",
  "typename": "stringShareGroupMetadataValue",
      "validVersions": "0",
  "aboutflexibleVersions": "0+"The,
 topic name."fields": },[
      { "name": "NumPartitionsEpoch", "versionstype": "0+int32", "typeversions": "int320+",
        "about": "Deprecated: this field is not used after 4.0. The number of partitions of the topic." }, <-- deprecated field
      { The group epoch." },
    { "name": "PartitionMetadataAllTopicHash", "versions": "0+", "type": "[]PartitionMetadataint64",
      "default": 0, "abouttaggedVersions": "0+", "tag"Deprecated: this0,
 field is not used after 4.0. Partitions mapped to a set "about": "The hash of racks.all If the rack information is unavailable for alltopics in 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
    ]}
  ]
}

ShareGroupPartitionMetadataValue

Add a new tagged field TopicHash.

{
  "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
    ]}
  ]
}

Proposed Changes

Topic Hash Map Cache in Coordinator

Different groups may subscribe to same topics. With topic hash map cache in the coordinator, it can avoid recalculation of same topic hash for different groups. The topic hash value should represent topic id, name, number of partitions, and partition racks.

Initial a topic hash in the cache

The coordinator only cares topics which are subscribed by groups. When a coordinator starts, it replays all records in __consumer_offsets. When replaying ConsumerGroupMemberMetadtaValue / ShareGroupMemberMetadataValue, the group can get subscribed topic names from previous state. In the next group heartbeat, the group can rely on subscribed topic names to compute topic hash and keep it in the cache, so other groups don't need to recompute it.

Remove a topic hash in the cache

If a topic is not subscribed by any group, the coordinator doesn't need to maintain the hash value. When replaying ConsumerGroupMemberMetadtaValue / ShareGroupMemberMetadataValue records, the coordinator adds a new topic to groupsByTopics if the topic is absent. If a group unsubscribes a topic, the group is removed from groupsByTopics. If there is no group under a topic, the topic will be removed from groupsByTopics. The group coordinator can leverage same mechanism to remove the topic hash.

Renew a topic hash in the cache

When there is a new metadata image, it contains changed topics and deleted topics in metadata delta. For both cases, the coordinator remove topic hash from the cache. The value will be recomputed when in the next consumer group heartbeat. The lazy evaluation is more efficient. If there is no further group heartbeat, the new topic hash value is useless, so we can just compute it in group heartbeat.

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();
            }
        });
    }
}

Subscribed Topic Hash Map in Group

To compute a new subscribed topic hash map, a group can leverage cache in the coordinator. If a topic hash is calculated by another group, we can just reuse it from cache.

Topic Hash Function

A topic hash represent topic id, name, number of partitions, and partition racks. When restarts the coordinator, the topic hash will be recomputed and keep in topic hash map. To avoid useless rebalance, the hash function should return same value for same data, even if it runs on different JDKs or partition racks have different order. For different JDKs, the KIP will use Murmur3 to compute the hash value. The function is already in org.apache.kafka.streams.state.internals. We will move it to org.apache.kafka.common.hash for common usage. For partition racks with different order, we will compute hash for each value in it and sum as a result.

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;
    }
}

SubscribedTopicDescriberImpl

group." } <-- new field
  ]
}

Proposed Changes

Topic Hash Map Cache in Coordinator

Different groups may subscribe to same topics. With topic hash map cache in the coordinator, it can avoid recalculation of same topic hash for different groups. The topic hash value should represent topic id, name, number of partitions, and partition racks.

Add a new topic hash in the cache

When the coordinator initializes, the topic hash map cache is empty. After receiving first consumer group heartbeat, the coordinator calculates subscribed topic hash, so it doesn't waste memory to store unsubscribed topic hash.

Renew a topic hash in the cache

When there is a new metadata image, it contains changed topics and deleted topics in metadata delta. For both cases, the coordinator remove topic hash from the cache. In the next consumer group heartbeat, the value will be recomputed and stored to the cache. The lazy evaluation is more efficient, because it guarantees that the coordinator only calculate topic hash which is in use.

Example: a new partition in a topic

When a new partition is added to a topic, the coordinator receives a new MetadataDelta. It contains the topic in TopicsDelta#changedTopics. The coordinator will clean up the topic hash. In the next group heartbeat, the coordinator detects the topic hash is empty and recalculate it.

Example: the racks of a partition change

The rack of a partition is from "broker.rack" configuration. This config can only be changed after the broker reboots. When the broker is stopped, the broker id is removed from PartitionRegistration#replicas. This is reflected in TopicDelta#partitionChanges. The renew process is similar to above. A new MetadataDelta contains the topic in TopicsDelta#changedTopics and the coordinator remove the topic hash after it receives the change. In the next group heartbeat, the coordinator re-compute the value.

All subscribed topic hash in Group

The group doesn't need to store subscribed topic metadata, because the coordinator doesn't need the value to detect a rebalance. After this KIP, the group gets all subscribed topic hash from the cache in coordinator and sums them as a single hash. The final hash will be stored with a new group epoch in ConsumerGroupMetadataValue / ShareGroupMetadataValue.

Topic Hash Function

A topic hash represents topic id, name, number of partitions, and partition racks. To avoid useless rebalance, the hash function should return same value for same data, even if it runs on different JDKs or partition racks have different order. For different JDKs, the KIP will use Murmur3 to compute the hash value. The function is already in org.apache.kafka.streams.state.internals. We will move it to org.apache.kafka.common.hash for common usage. For partition racks with different order, we will compute hash for each value in it and sum as a result.

SubscribedTopicDescriberImpl

The group doesn't store topic metadataThe topic metadata doesn't have number of partitions and rack data, so the subscribedTopicDescriberImpl can't rely on it to return numPartitions and racksForPartition. This KIP adds MetadataImage to SubscribedTopicDescriberImpl constructorchanges the constructor from Map<Uuid, TopicMetadata> to Set<Uuid> and MetadataImage, so it can use Map<Uuid, TopicMetadata> Set<Uuid> to check which topic is subscribed and use MetadataImage to return numPartitions and racksForPartition.

Compatibility, Deprecation, and Migration Plan

ConsumerGroupPartitionMetadataValue / ShareGroupPartitionMetadataValue

There is new field in ConsumerGroupPartitionMetadataValue and ShareGroupPartitionMetadataValuetwo records will be removed. For compatibility, adding it as tagged fieldwe should keep GroupMetadataManager#replay functions, so old brokers can deserialize itthe coordinator can read old data.

Test Plan

• Unit test for the topic hash function. The hash function should ignore partition racks order and give a same value if the set is no difference.
• Unit test for GroupMetadataManager. Test only topic id, name, number of partition, and partition racks change can make a rebalance. Other data change should keep same assignment.
• Integration test between coordinator and consumer. Use admin client to update topic partition and check consumer group get a new assignment. Restart a broker to change rack and check consumer group get a new assignment. Use a consumer group to subscribe regex pattern and use admin client to add a new pattern matched topic, check the consumer group get a new assignment.

Rejected Alternatives

Single Hash

...

Check Topic Delta to Trigger a Rebalance

...