Page History

import com.google.common.hash.HashCode;
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 {
    // ...

    /**
     * Compute metadata hash based on the current subscription info.
     *
     * @param subscribedTopicNames Map of topic names to the number of subscribers.
     * @param metadataImage        The current metadata for all available topics.
     * @param topicHashCache       The cache of topic hashes.
     */
    public long computeMetadataHash(
        Map<String, SubscriptionCount> subscribedTopicNames,
        MetadataImage metadataImage,
        Map<String, Long> topicHashCache
    ) {
        TopicsImage topicsImage = metadataImage.topics();
        List<HashCode> hashCodes = subscribedTopicNames.keySet().stream()
            .filter(topicName -> topicsImage.getTopic(topicName) != null)
            .map(topicName -> HashCode.fromLong(
                topicHashCache.computeIfAbsent(
                    topicName,
                    key -> computeTopicHash(topicsImage.getTopic(topicName), metadataImage.cluster())
                )
            ))
            .toList();
        return hashCodes.isEmpty() ? 0 : Hashing.combineUnordered(
            hashCodes
        ).asLong();
    }
}

import com.google.common.hash.HashCode;
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();
    }
}