Page History

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

    /**
     * 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.
     */
    private long computeTopicHash(TopicImage topicImage, ClusterImage clusterImage) {
        long result = topicImage.id().hashCode();
        result = 63 * result + Murmur3.hash64(topicImage.name().getBytes(StandardCharsets.UTF_8));
        result = 63 * result + Murmur3.hash64(topicImage.partitions().size());
        result = 63 * result + topicImage.partitions().entrySet().stream().mapToLong(
            entry -> {
                PartitionRegistration partitionRegistration = entry.getValue();
                long partitionHash = Murmur3.hash64(entry.getKey());
                Set<String> racks = Arrays.stream(partitionRegistration.replicas)
                    .mapToObj(clusterImage::broker)
                    .map(BrokerRegistration::rack)
                    .filter(Optional::isPresent)
                    .map(Optional::get)
                    .collect(Collectors.toSet());
                return 63 * partitionHash + racks.stream().mapToLong(
                    rack -> Murmur3.hash64(rack.getBytes(StandardCharsets.UTF_8))
                ).reduce(0L, Long::sum);
            }
        ).reduce(0L, Long::sum);
        return result;
    } 
}