Page History

...

When there is a new metadata image, it contains changed topics and deleted topics in metadata delta. For changed topicsboth cases, the coordinator updates new hash if it is subscribed by at least one group. This avoids useless calculation. For deleted topics, the coordinator remove it from the cacheremove topic hash from the cache. The value will be recomputed when in the next consumer group heartbeat. The lazy evaluation is more efficient. Also, if there is no further group heartbeat, the new topic hash value is useless, so we can just compute it in group heartbeat.

Code Block

language	java
linenumbers	true

public class GroupMetadataManager {
    // ...

    /**
     * The topic hash value by topic name.
     */
    private final Map<String, Long> topicHashCache; 

    /**
     * Subscribes a group to a topic.
     *
     * @param groupId   The group id.
     * @param topicName The topic name.
     */
    private void subscribeGroupToTopic(
        String groupId,
        String topicName
    ) {
        groupsByTopics
            .computeIfAbsent(topicName, __ -> {
                topicHashes.put(topicName, computeTopicHash(topicName));
                return new TimelineHashSet<>(snapshotRegistry, 1);
            })
            .add(groupId);
    }

    /**
     * Unsubscribes a group from a topic.
     *
     * @param groupId   The group id.
     * @param topicName The topic name.
     */
    private void unsubscribeGroupFromTopic(
        String groupId,
        String topicName
    ) {
        groupsByTopics.computeIfPresent(topicName, (__, groupIds) -> {
            groupIds.remove(groupId);
            if (groups.isEmpty()) {
                topicHashCache.remove(topicName);
                return null;
            }
            return groupIds;
        });
    }

    /**
     * A new metadata image is available.
     *
     * @param newImage  The new metadata image.
     * @param delta     The delta image.
     */
    public void onNewMetadataImage(MetadataImage newImage, MetadataDelta delta) {
        metadataImage = newImage;

        // Notify all the groups subscribed to the created, updated or
        // deleted topics.
        Optional.ofNullable(delta.topicsDelta()).ifPresent(topicsDelta -> {
            Set<String> allGroupIds = new HashSet<>();
            topicsDelta.changedTopics().forEach((topicId, topicDelta) -> {
                String topicName = topicDelta.name();
                topicHashCache.remove(topicName);
     Set<String> groupIds =            allGroupIds.addAll(groupsSubscribedToTopic(topicName));
            });
           if topicsDelta.deletedTopicIds(!groupIds).isEmpty())forEach(topicId -> {
                TopicImage topicImage =  topicHashCachedelta.put(topicName, computeTopicHash(topicName)image().topics().getTopic(topicId);
                String topicName =  allGroupIdstopicImage.addAllname(groupIds);
                topicHashCache.remove(topicName);
              }   allGroupIds.addAll(groupsSubscribedToTopic(topicName));
                   });
            topicsDelta.deletedTopicIds()allGroupIds.forEach(topicIdgroupId -> {
                TopicImageGroup topicImagegroup = deltagroups.imageget().topics().getTopic(topicId)groupId);
                Stringif topicName(group != topicImage.namenull && (group.type();
 == CONSUMER || group.type() == SHARE)) {
         Set<String> groupIds = groupsSubscribedToTopic(topicName);
                if (!groupIds.isEmpty()) {
                    topicHashCache.remove(topicName((ModernGroup<?>) group).requestMetadataRefresh();
                    allGroupIds.addAll(groupIds);}
                });
            });
            allGroupIds.forEach(groupId -> {
                Group group = groups.get(groupId);
                if (group != null && (group.type() == CONSUMER || group.type() == SHARE)) {
                    ((ModernGroup<?>) group).requestMetadataRefresh();}
}

Subscribed Topic Hash Map in Group

A group has hard and soft state. The hard state can only be updated by replaying records. The subscribed topic hash map is hard state, so it will be updated when replaying ConsumerGroupPartitionMetadataValue and ShareGroupPartitionMetadataValue.

To compute a new subscribed topic hash map, a group can leverage cache in the coordinator, so it doesn't need to recompute it.

Code Block

language	java
linenumbers	true

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   });
    }
}

Subscribed Topic Hash Map in Group

A group has hard and soft state. The hard state can only be updated by replaying records. The subscribed topic hash map is hard state, so it will be updated when replaying ConsumerGroupPartitionMetadataValue and ShareGroupPartitionMetadataValue.

To compute a new subscribed topic hash map, a group can leverage cache in the coordinator, so it doesn't need to recompute it.

Code Block

language	java
linenumbers	true

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

    /**
     * The hash with each subscribed topic name.consumer group is subscribed to.
     */
    public Map<String, Long> subscriptionTopicHash() {
        return Collections.unmodifiableMap(subscriptionTopicHash);
    }


    /**
     */
    protected final TimelineHashMap<String, Long> subscriptionTopicHash;

 Updates the subscription topic hash. This replaces the previous one.
     /**
     * @return@param AnsubscriptionTopicHash immutableThe Map ofnew subscription topic hash for.
     */
    public void setSubscriptionTopicHash(
   each topic that the consumer groupMap<String, is subscribed to.Long> subscriptionTopicHash
    ) */{
     public Map<String, Long> this.subscriptionTopicHash.clear() {;
        return Collections.unmodifiableMapthis.subscriptionTopicHash.putAll(subscriptionTopicHash);
    }


    /**
     * UpdatesComputes the subscription topic hash. Thisbased replaceson the previouscurrent subscription oneinfo.
     * 
     * @param subscriptionTopicHashsubscribedTopicNames The newMap subscriptionof topic hashnames to the number of subscribers.
     */
 @param topicHashCache  public void setSubscriptionTopicHash(
    The current topic hash Map<String,cache Long>from subscriptionTopicHashcoordinator.
    ) {*
     * @return  this.subscriptionTopicHash.clear();
        this.subscriptionTopicHash.putAll(subscriptionTopicHash);
    }

    /**An immutable map of subscription topic hash for each topic that the consumer group is subscribed to.
     */
 Computes the subscription topicpublic hashMap<String, basedLong> oncomputeSubscriptionTopicHash(
 the current subscription info.
    Map<String, *Integer> subscribedTopicNames,
       * @paramTopicsImage subscribedTopicNamestopicsImage,
  Map of topic names to the numberMap<String, ofLong> subscribers.topicHashCache
    ) *{
 @param topicHashCache      // Create The currentthe topic hash cachefor each fromsubscribed coordinatortopic.
       * Map<String, Long> newSubscriptionTopicHash = new HashMap<>();
     *  @return An immutable map of subscription topic hash for each topic that the consumer group is subscribed to.
 subscribedTopicNames.forEach((topicName, count) -> {
            TopicImage topicImage = topicsImage.getTopic(topicName);
         */
   if public Map<String, Long> computeSubscriptionTopicHash(
        Map<String, Integer> subscribedTopicNames,
  (topicImage != null) {
                TopicMetadata topicMetadata = new TopicMetadata(
      Map<String, Long> topicHashCache
    ) {
        // Create the topic hash for each subscribed topic.
   topicImage.id(),
           Map<String, Long> newSubscriptionTopicHash = new HashMap<>();
        subscribedTopicNamestopicImage.forEach((topicName, count) -> {
name(),
              Long topicHash = topicHashCache.get(topicName);
       topicImage.partitions().size()
     if (topicHash != null) {
       );
                newSubscriptionTopicHash.put newSubscriptionTopicHash.put(topicName, cache.computeIfAbsent(topicName, topicHash__ -> topicMetadata.hashCode()));
            }
        });
        return Collections.unmodifiableMap(newSubscriptionTopicHash);
    }
}

...

Space shortcuts

Child pages

Versions Compared

Old Version 4

New Version 5

Key

Subscribed Topic Hash Map in Group

Subscribed Topic Hash Map in Group