Page History

...

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 the metadata is expired, the group coordinator computes new subscribed topic metadata and compare with current one. 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 subscribed topic metadata and only use use a subscribed topic hash map. Each topic has a hash value to represent it. There are two server side conditions can trigger a new rebalance, so the topic hash needs to reflect them:

...

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

...

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. A topic will be removed from groupsByTopics if there is no group subscribe to it. The group coordinator can remove the topic hash as well, so it can leverage same mechanism to remove a topic hash.

...

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. Also, if If there is no further group heartbeat, the new topic hash value is useless, so we can just compute it in group heartbeat.

...

To compute a new subscribed topic hash map, a group can leverage cache in the coordinator, so it doesn't need to recompute it. If a topic hash is calculated by another group, we can just reuse it from cache.

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,
        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)));
            }
        });
        return Collections.unmodifiableMap(newSubscriptionTopicHash);
    }
}

...