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

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
     * @param topicImage   The topic image.
  each topic that the* consumer@param groupclusterImage isThe subscribedcluster toimage.
     */
    private long publiccomputeTopicHash(TopicImage Map<StringtopicImage, Long>ClusterImage subscriptionTopicHash(clusterImage) {
        return Collections.unmodifiableMap(subscriptionTopicHashlong result = topicImage.id().hashCode();
     }


    /**   result = 63 * result + Murmur3.hash64(topicImage.name().getBytes(StandardCharsets.UTF_8));
     * Updates the subscriptionresult topic= hash.63 This* replacesresult the previous one.
+ Murmur3.hash64(topicImage.partitions().size());
      *
  result = 63 * @paramresult subscriptionTopicHash The new subscription topic hash.
     */+ topicImage.partitions().entrySet().stream().mapToLong(
    public void setSubscriptionTopicHash(
        Map<String,entry Long>-> subscriptionTopicHash{
    ) {
        this.subscriptionTopicHash.clear();
    PartitionRegistration partitionRegistration =  this.subscriptionTopicHash.putAll(subscriptionTopicHashentry.getValue();
    }

    /**
     * Computes the subscriptionlong topicpartitionHash hash based on the current subscription info.= Murmur3.hash64(entry.getKey());
     * 
     * @param subscribedTopicNames  Map ofSet<String> topicracks names to the number of subscribers.= Arrays.stream(partitionRegistration.replicas)
     * @param topicHashCache        The current topic hash cache from coordinator.mapToObj(clusterImage::broker)
      *
     * @return An immutable map of subscription topic hash for each topic that the consumer group is subscribed to.
 .map(BrokerRegistration::rack)
             */
    public Map<String, Long> computeSubscriptionTopicHash(.filter(Optional::isPresent)
        Map<String,  Integer> subscribedTopicNames,
        TopicsImage topicsImage, .map(Optional::get)
        Map<String, Long> topicHashCache
    ) {
        // Create the topic hash for each subscribed topic.
.collect(Collectors.toSet());
                Map<String,return Long>63 newSubscriptionTopicHash* =partitionHash new+ HashMap<>racks.stream();.mapToLong(
        subscribedTopicNames.forEach((topicName, count) -> {
            TopicImage topicImage = topicsImage.getTopic(topicName);
            rack -> Murmur3.hash64(rack.getBytes(StandardCharsets.UTF_8))
             if (topicImage != null) {
                newSubscriptionTopicHash.put(topicName, cache.computeIfAbsent(topicName, t -> computeTopicHash(topicImage))).reduce(0L, Long::sum);
            }
         }).reduce(0L, Long::sum);
        return Collections.unmodifiableMap(newSubscriptionTopicHash)result;
    }
 }