Page History

...

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

import com.google.common.hash.HashFunction;
import com.google.common.hash.Hasher;
import com.google.common.hash.Hashing;

public abstract class ModernGroup<T extends ModernGroupMember> implements Group {
    // ...

    /**
     * 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) {
        HashFunction hf = Hashing.murmur3_128();
        Hasher topicHasher = hf.newHasher()
            .putByte((byte) 0) // magic byte
            .putLong(topicImage.id().hashCode()) // topic Id
            .putString(topicImage.name(), StandardCharsets.UTF_8) // topic name
            .putLong(topicImage.partitions().size()); // number of partitions

        topicImage.partitions().entrySet().stream().sorted(Map.Entry.comparingByKey()).forEach(entry -> {
            topicHasher.putInt(entry.getKey()); // partition id
            Arrays.stream(entry.getValue().replicas)
                .mapToObj(clusterImage::broker)
                .filter(Objects::nonNull)
                .map(BrokerRegistration::rack)
                .filter(Optional::isPresent)
                .map(Optional::get)
                .sorted()
                .forEach(rack -> topicHasher.putString(rack, StandardCharsets.UTF_8)); // sorted racks
        });
        return topicHasher.hash().asLong();
    }
}

...