...

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.

...

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. 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. Any A single bit change can get a different hash value.

...

Compatibility, Deprecation, and Migration Plan

ConsumerGroupPartitionMetadataValue

...

This record There two records will be deprecated. For compatibility, we should keep GroupMetadataManager#replay functionsthe replay function, so the coordinator can read old data.

...

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 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. At the first time, the coordinator writes a tombstone ConsumerGroupPartitionMetadataValue, so old data can be removed after log compaction.

Code Block
ConsumerGroupMemberMetadataValue ConsumerGroupPartitionMetadataValue ConsumerGroupMetadataValue (no MetadataHash field) ConsumerGroupTargetAssignmentMemberValue ConsumerGroupTargetAssignmentMetadataValue ConsumerGroupCurrentMemberAssignmentValue

...

• 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.
• 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

...