Page History

public class GroupMetadataManager {
    // ...

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

    /**
     * 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 allInitialize the groupslast subscribedoffset toif theit created,was updatednot oryet.
        if (lastMetadataImageWithNewTopics // deleted topics.
== -1L) {
            lastMetadataImageWithNewTopics = OptionalmetadataImage.ofNullableprovenance(delta).topicsDelta()).ifPresent(topicsDelta -> {
    lastContainedOffset();
        }

        TopicsDelta topicsDelta = delta.topicsDelta();
        if (topicsDelta == null) return;

        // Updated the last offset of the image with newly created topics. This is used to
        // trigger a refresh of all the regular expressions when topics are created. Note
        // that we don't trigger a refresh when topics are deleted. Those are removed from
        // the subscription metadata (and the assignment) via the above mechanism. The
        // resolved regular expressions are cleaned up on the next refresh.
        if (!topicsDelta.createdTopicIds().isEmpty()) {
            lastMetadataImageWithNewTopics = metadataImage.provenance().lastContainedOffset();
        }

        // Notify all the groups subscribed to the created, updated or
        // deleted topics.
        Set<String> allGroupIds = new HashSet<>();
            topicsDelta.changedTopics().forEach((topicId, topicDelta) -> {
                String topicName = topicDelta.name();
            // trigger recalculate topic hash in next consumer group heartbeat
            topicHashCache.remove(topicName);
                    allGroupIdsallGroupIds.addAll(groupsSubscribedToTopic(topicName));
              });
            topicsDelta.deletedTopicIds().forEach(topicId -> {
     {
            TopicImage topicImage = delta.image().topics().getTopic(topicId);
                String topicName = topicImage.name();
                    topicHashCachetopicHashCache.remove(topicName);
                    allGroupIdsallGroupIds.addAll(groupsSubscribedToTopic(topicName));
              });
            allGroupIds.forEach(groupId -> {
                Group group = groups.get(groupId);
                if (group != null && (group.type() == CONSUMER || group.type() == SHARE)) {
                    ((ModernGroup<?>) group).requestMetadataRefresh();
                }
            });
        });
    }
}