...

KIP-848 introduces a next generation of consumer rebalance protocol which supports rack-aware partition assignment. In the first implement, the group coordinator computes subscribed topic metadata which contains topic UUID, name, number of partition, and rack set of each partition. If result is difference, the group coordinator bumps group epoch and calculates new target assignment which means it triggers a rebalance. However, the rack set of each partition takes too much space. In KAFKA-17578, a real case of memory usage shows that in a group with 500 members and 2K topic partitions,  partition racks account for 79% of whole ConsumerGroup object. This KIP will get rid of TopicMetadata and introduce a new way to detect only use a hash value to represent it. There are two server side rebalance conditionsconditions can trigger a new rebalance, so the topic hash needs to reflect them:

• A topic with a new partition.
• A topic partition has rack change.

Public Interfaces

ConsumerGroupPartitionMetadataValue

...

Since topic metadata will be deprecated, we don't need to store it in ConsumerGroupPartitionMetadataValue, either. Both ConsumerGroupPartitionMetadataValue from KIP-848 and ShareGroupPartitionMetadataValue from KIP-932 will be deprecated.

Proposed Changes

New Metadata Image

Without calculating subscribe topic metadata, the group coordinator needs to detect a topic change when it receives new MetadataImage and MetadataDelta. The MetadataDelta contains a topic change as TopicDelta.

New partitions can be detected by TopicDelta#newPartitions.

For rack change, it happens when broker.rack is updated. The broker.rack value is read-only config. It can only be changed when a broker restarts. If a broker stops, it will be removed from topic partition replicas. If it starts, it will be added to topic partition replicas again. In TopicDelta, each partition change is in PartitionRegistration. The rack change can be detected by PartitionRegistration#addingReplicas and PartitionRegistration#removingReplicas.

After catching all rebalance conditions, set a value in the group to indicate that the group need rebalance in the next group heartbeat.

ModernGroup

Add a new boolean attribute triggerRebalanceOnNextHeartbeat to the group with the initial value false. When the rebalance condition is matched, set the value to true. When a rebalance is handled, set the value to false. This will replace function ModernGroup#requestMetadataRefresh.

...

Add a new field TopicHash. 

Code Block
title ConsumerGroupPartitionMetadataValue linenumbers true
{ "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." } ]} ] }

ShareGroupPartitionMetadataValue

Add a new 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": "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." }
    ]}
  ]
}

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 recalculate same topic hash for different groups. The topic hash value should represent topic id, name, number of partitions, and partition racks.

Initialize the topic hash map cache in coordinator

The coordinator only cares topics which are subscribed by groups. If a topic is not subscribed by any group, the coordinator doesn't need to maintain the hash value. When a coordinator starts, it replays all records in __consumer_offsets. When a coordinator replays ConsumerGroupMemberMetadtaValue / ShareGroupMemberMetadataValue, it adds a new topic to groupsByTopics if the topic is absent. The group coordinator can leverage it to add a new topic hash.

Remove topic hash from the cache in coordinator

If a topic was subscribed by a group, but there is no group subscribes it now, it can be removed from the cache. When there is no group subscribe to the topic, it will be removed from groupsByTopics. The group coordinator can leverage it to remove a topic hash.

Update topic hash in the cache in coordinator

When there is a new metadata image, it contains changed topics and deleted topics in metadata delta. For changed topics, the coordinator updates new hash if it is subscribed by at least one group. This avoids useless calculation. For deleted topics, the coordinator remove it from the cache.

public class GroupMetadataManager {
    // ...

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

    /**
     * Subscribes a group to a topic.
     *
     * @param groupId   The group id.
     * @param topicName The topic name.
     */
    private void subscribeGroupToTopic(
        String groupId,
        String topicName
    ) {
        groupsByTopics
            .computeIfAbsent(topicName, __ -> {
                topicHashes.put(topicName, computeTopicHash(topicName));
                return new TimelineHashSet<>(snapshotRegistry, 1);
            })
            .add(groupId);
    }

    /**
     * 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()) {
                topicHashes.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();
                Set<String> groupIds = groupsSubscribedToTopic(topicName);
                if (!groupIds.isEmpty()) {
                    topicHashes.put(topicName, computeTopicHash(topicName));
                    allGroupIds.addAll(groupIds);
                }
            });
            topicsDelta.deletedTopicIds().forEach(topicId -> {
                TopicImage topicImage = delta.image().topics().getTopic(topicId);
                String topicName = topicImage.name();
                Set<String> groupIds = groupsSubscribedToTopic(topicName);
                if (!groupIds.isEmpty()) {
                    topicHashes.remove(topicName);
                    allGroupIds.addAll(groupIds);
                }
            });
            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

A group has hard and soft state. The hard state can only be updated by replaying records. The subscribed topic hash map is hard state, so it will be updated when replaying ConsumerGroupPartitionMetadataValue and ShareGroupPartitionMetadataValue. To compute a new subscribed topic hash map, a group can leverage cache in the coordinator, so it doesn't need to recompute it.

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,
        Map<String, Long> topicHashCache
    ) {
        // Create the topic metadata for each subscribed topic.
        Map<String, Long> newSubscriptionTopicHash = new HashMap<>();
        subscribedTopicNames.forEach((topicName, count) -> {
            Long topicHash = topicHashCache.get(topicName);
            if (topicHash != null) {
                newSubscriptionTopicHash.put(topicName, topicHash);
            }
        });
        return Collections.unmodifiableMap(newSubscriptionTopicHash);
    }
}

SubscribedTopicDescribeImpl

...

Compatibility, Deprecation, and Migration Plan

GroupMetadataManager

ConsumerGroupPartitionMetadataValue / ShareGroupPartitionMetadataValue

There is new field in ConsumerGroupPartitionMetadataValue and ShareGroupPartitionMetadataValue. For compatibility, adding it as tagged field, so old brokers can deserialize To replay deprecated records (ConsumerGroupPartitionMetadataValue and ShareGroupPartitionMetadataValue) in __consumer_offsets, the GroupMetadataManager has to keep replay functions and do nothing in it.

Test Plan

Describe in few sentences how the KIP will be tested. We are mostly interested in system tests (since unit-tests are specific to implementation details). How will we know that the implementation works as expected? How will we know nothing broke?

Rejected Alternatives

...