Status

Current state: Under Discussion Accepted

Discussion thread: https://lists.apache.org/thread/l8ko353v3nn1blgymsty895x6c98oxlx

Vote thread: https://lists.apache.org/thread/j90htmphjk783p697jjfg1xt8thmy33p

JIRA:

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

KIP-848 introduces a next-generation

...

consumer rebalance protocol

...

that supports rack-aware partition assignment. In

...

its initial implementation, the group coordinator computes subscribed topic metadata, which

...

includes the topic UUID, name, number of

...

partitions, and the rack set

...

for each partition.

...

When this metadata expires, the group coordinator

...

recalculates the subscribed topic metadata and

...

compares it to the current version. If there are differences, the group coordinator

...

increments the group epoch and

...

generates a new target assignment

...

, effectively triggering a rebalance. However, the rack set

...

for each partition

...

consumes significant memory. In KAFKA-17578, a real-world case

...

shows that

...

for a group with 500 members and

...

2,000 topic partitions,

...

partition

...

rack data accounts for 79% of

...

the total memory used by the ConsumerGroup object.

...

The rack-aware rebalance was removed in 4.0. This KIP introduces another way to support the feature and targets at 4.1

This KIP proposes removing the number of partitions and rack set details from the metadata and replacing them with a subscribed topic hash

...

. Each topic

...

is assigned a unique hash value.

...

The topic hash must account for two server-side conditions that can trigger a

...

rebalance

...

:

• A topic with a new partition.

• A topic partition

  has

  undergoes a rack change. Each topic partition has multiple replicas.

  Each replica is stored on a broker.

  The rack value is derived from the broker's broker.rack

  config. It's a read only config and can only be changed when

  configuration, which is read-only and only changes when the broker restarts. If a broker stops,

  a related replica will be

  its associated replica is removed from the topic partitions.

  If

  Conversely, when a broker starts,

  it will be

  its replica is added to the topic partitions.

Compare the subscribed topic metadata, a single hash of the subscribed topic metadata, and the subscribed topic hash map. The subscribed topic metadata requires the most storage but the least computation. A single hash of the subscribed topic metadata requires the least storage but the most computation. A balanced approach would aim to optimize both computation and storage usage. This KIP uses a hash value for each topic, so a new hash only needs to be computed when there is a change to the topic. Each topic only stores a hash value, so we don’t need to store all detailed information.

Public Interfaces

ConsumerGroupPartitionMetadataValue

Add a new field TopicHash. 

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

The second strategy still requires significant disk space to store redundant data. It stores a map of each subscribed topic hash. Even if only a single entry has changed, it results in storing a new map of topic hash. On the other hand, the third strategy heavily consumes CPU resources to recalculate the hash whenever a topic changes.

This KIP proposes combining the second and third strategies. The coordinator will maintain a cache to store individual topic hashes. When a topic changes, only its hash will be recalculated. Different groups can retrieve the topic hash from the cache, eliminating the need for the coordinator to recalculate the same hash multiple times. In the record, the coordinator will aggregate all subscribed topic hashes into a single hash per group and store it on disk. This approach minimizes disk usage while efficiently handling changes.

Public Interfaces

ConsumerGroupPartitionMetadataKey / ConsumerGroupPartitionMetadataValue

These types will be deprecated for backward compatibility.

StreamsGroupPartitionMetadataKey / StreamsGroupPartitionMetadataValue / ShareGroupPartitionMetadataKey / ShareGroupPartitionMetadataValue

These types will be deleted, because Stream / Share group coordinator has not been GA before 4.1.

ConsumerGroupMetadataValue

Add a new field "MetadataHash". Set it as tagged field for backward compatibility.

{
  "type": "data",
 

{
  "type": "data",
  "name": "ConsumerGroupPartitionMetadataValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "TopicsConsumerGroupMetadataValue",
  "versionsvalidVersions": "0+",
  "typeflexibleVersions": "[]TopicMetadata0+",
      "about": "The list of topic metadata.", "fields": [
      { "name": "TopicIdEpoch", "versions": "0+", "type": "uuidint32",
        "about": "The topicgroup idepoch." },
      { "name": "TopicNameMetadataHash", "versions": "0+", "type": "stringint64",
      "default": 0, "abouttaggedVersions": "The topic name." },
      { "name": "NumPartitions", "versions0+", "tag": "0+", "type": "int32",
        "about": "The numberhash of all partitionstopics ofin the topicgroup." },
 <-- new field
   { "]
}

ShareGroupMetadataValue

Add a new field "MetadataHash".

{
  "type": "data",
  "name": "PartitionMetadataShareGroupMetadataValue",
  "versionsvalidVersions": "0+",
  "typeflexibleVersions": "[]PartitionMetadata0+",
  "fields": [
    { "aboutname": "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", "Epoch", "type": "int32", "versions": "0+",
      "about": "The group epoch." },
    { "name": "MetadataHash", "versions": "0+", "type": "int32int64",
            "about": "The hash partitionof all topics in the group." } <-- new field
  ]
}

StreamsGroupMetadataValue

Add a new field "MetadataHash".

{
  "typenumber." },
          { "name": "Racksdata",
  "versionsname": "0+StreamsGroupMetadataValue",
  "typevalidVersions": "[]string0",
            "about"flexibleVersions": "The set of racks that the partition is mapped to." }0+",
      ]},
  "fields": [
    { "name": "TopicHashEpoch", "versions": "0+", "type": "int64int32",
        "defaultabout": 0, "taggedVersions"The group epoch." },
    { "name": "MetadataHash", "versions": "0+", "tagtype": 0"int64",
        "about": "The hash value of theall topictopics id,in name, number of partitions, and partition racksthe group." }
 <-- new  ]}field
  ]
}

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

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 the 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

There are two cases about rack change. One is users alter partition reassignment. Another is "broker.rack" configuration change. The configuration is read only and needs to be changed after the broker reboots. For both cases, if the replica is ISR on a reassigned/rebooted broker, the change is reflected in PartitionRegistration#isr. This is also included 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.

The PartitionRegistration also contains other changes like leader / elr. Every change will make the topic hash recomputation. It's no harm, because the hash only reflects change about uuid / name / number of partition / racks of partitions. If non of these fields change, the final hash will be the same, so it doesn't bump the group epoch or trigger a rebalance.

Remove a topic hash in the cache

In the coordinator, it maintains each topic is subscribed by which groups. When a group unsubscribes, the coordinator checks whether a topic is subscribed by any group. If it's not, the topic hash is removed from the cache.

A single hash of all subscribed topic in Group

In 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 / StreamGroupMetadataValue. A simple sum of each topic hash may have collision For example, topicA gets hash value "a" and topicB gets "b". After some operations (add partition / rack change), topicA gets hash value "b" and topicB gets "a". In this case, a simple sum cannot trigger a rebalance. To ensure an avalanche effect, the combined hash function should sort topics by name and each hash value multiplied by the position value.

Another case is about regular expression. For example, a consumer subscribe regular expression like "topic*". At T1, matched topics are topicA with hash value "a" and topicB with hash value "b". At T2, topicA and topicB are removed. The topicC and topicD are added and have same hash value "a" and "b". This case is not covered by this KIP because the group hash value cannot trigger the rebalance. In 4.0, there is another function executes regular expression and check whether topic result is different. If yes, it bumps the group epoch and calculates new assignment, so this KIP doesn't need to handle this case.

Topic Hash Function

A topic hash represents 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;

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

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

SubscribedTopicDescribeImpl

Since the group coordinator doesn't compute subscribed topic metadata, the SubscribedTopicDescribeImpl can't use Map<Uuid, TopicMetadata> to number of partition and rack set. We will use MetadataImage as input to replace it.

Compatibility, Deprecation, and Migration Plan

ConsumerGroupPartitionMetadataValue / ShareGroupPartitionMetadataValue

...

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. For partition racks with different order, we will compute hash for each value in it and sum as a result. We will also set the first byte as magic byte to represent hash version. The Murmur3 uses bit operation to ensure avalanche effect. A single bit change can get a different hash value. The hash function should follow the order to combine different fields:

• magic byte (byte)
• topic id (long)
• topic name (string)
• partition size (long)
• sorted partition by id
  • partition id (int)
  • sorted racks (string)

SubscribedTopicDescriberImpl

The group doesn't store TopicMetadata, so the subscribedTopicDescriberImpl can't rely on it to return numPartitions and racksForPartition. This KIP changes the constructor from Map<Uuid, TopicMetadata> to Set<Uuid> and MetadataImage, so it can use Set<Uuid> to check which topic is subscribed and use MetadataImage to return numPartitions and racksForPartition.

Compatibility, Deprecation, and Migration Plan

ConsumerGroupPartitionMetadataValue

This record will be deprecated. For compatibility, we should keep the replay function, so the coordinator can read old data.

Upgrade

When a coordinator upgrades, it initials with old records like following. It's no harm for the coordinator to read ConsumerGroupPartitionMetadataValue, because the coordinator still keeps the replay function. When it reads the record, it creates a group if the value is not null or does nothing if value is null. New logic will try read MetadataHash field. Although there is no MetadataHash field in the old ConsumerGroupMetadataValue record, it's tagged field with default value 0, so the coordinator sets a group with an initial value 0. When the coordinator receives the first heartbeat, it calculates a new MetadataHash and triggers a group epoch bump without a rebalance, because assignors are sticky. The topology of the group will not change. At the first time, the coordinator writes a tombstone ConsumerGroupPartitionMetadataValue, so old data can be removed after log compaction.

ConsumerGroupMemberMetadataValue
ConsumerGroupPartitionMetadataValue
ConsumerGroupMetadataValue (no MetadataHash field)
ConsumerGroupTargetAssignmentMemberValue
ConsumerGroupTargetAssignmentMetadataValue
ConsumerGroupCurrentMemberAssignmentValue

Downgrade

When a coordinator downgrades, it initials with new records like following. The coordinator doesn't have logic to handle MetadataHash, so the field is ignored. It initials a group without subscription metadata. When the coordinator receives the first heartbeat, it calculates a new subscription metadata and finds the value is different. This triggers a group epoch bump without a rebalance, because assignors are sticky. The topology of the group will not change.

ConsumerGroupMemberMetadataValue
ConsumerGroupMetadataValue (with MetadataHash tagged field)
ConsumerGroupTargetAssignmentMemberValue
ConsumerGroupTargetAssignmentMetadataValue
ConsumerGroupCurrentMemberAssignmentValue

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

...

• Extend ConsumerGroupHeartbeatRequestTest to cover new feature: add a new partition or change rack can trigger a rebalance.

Rejected Alternatives

Check Topic Delta to Trigger a Rebalance

...

A new metadata image is computed by controller. An epoch can represent the version of TopicImage. If there is number of partitions or partition racks change, the controller bumps the epoch. When coordinator receives a new metadata image and there is a new topic epoch, it triggers a rebalance for a group. The group also can store topic epoch map in records, so it can know the difference if the coordinator restarts. This approach save cpu and storage resources and avoid the downside of "Check Topic Delta to Trigger a Rebalance". However, this approach mixes the group coordinator logic within the controllerTopic Delta to Trigger a Rebalance". However, this approach mixes the group coordinator logic within the controller.

Bump Group Version

Increase the group.version feature level to 2 to enable rack-aware rebalancing. However, there are two drawbacks to implementing this:

1. It requires maintaining lot of if-else block to determine whether to use subscription metadata or group hash.
2. When changing the group.version to 2, it triggers an epoch bump for all groups.