...

The coordinator only cares topics which are subscribed by groups. If a topic is not subscribed by any group, the coordinator doesn't need to maintain the hash value. When a coordinator starts, it replays all records in __consumer_offsets. When a coordinator replays replaying ConsumerGroupMemberMetadtaValue / ShareGroupMemberMetadataValue, it the coordinator adds a new topic to groupsByTopics if the topic is absent. The group coordinator can leverage it to add a new topic hash.

...

If a topic was subscribed by a group, but there is no group subscribes it now, it can be removed from the cache. When A topic will be removed from groupsByTopics if there is no group subscribe to the topic, it will be removed from groupsByTopics. The group coordinator can leverage it same mechanism to remove a topic hash.

Update topic hash in the cache in coordinator

...

public class GroupMetadataManager {
    // ...

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

    /**
     * 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()) {
                topicHashestopicHashCache.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();
                Set<String> groupIds = groupsSubscribedToTopic(topicName);
                if (!groupIds.isEmpty()) {
                    topicHashestopicHashCache.put(topicName, computeTopicHash(topicName));
                    allGroupIds.addAll(groupIds);
                }
            });
            topicsDelta.deletedTopicIds().forEach(topicId -> {
                TopicImage topicImage = delta.image().topics().getTopic(topicId);
                String topicName = topicImage.name();
                Set<String> groupIds = groupsSubscribedToTopic(topicName);
                if (!groupIds.isEmpty()) {
                    topicHashestopicHashCache.remove(topicName);
                    allGroupIds.addAll(groupIds);
                }
            });
            allGroupIds.forEach(groupId -> {
                Group group = groups.get(groupId);
                if (group != null && (group.type() == CONSUMER || group.type() == SHARE)) {
                    ((ModernGroup<?>) group).requestMetadataRefresh();
                }
            });
        });
    }
}

...

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.

...

Describe in few sentences how the KIP will be tested. We are mostly interested in system tests (since unit-tests are specific to implementation details). How will we know that the implementation works as expected? How will we know nothing broke?

Rejected Alternatives

Single Hash

Using a single hash to represent subscribed topic metadata can detect when to trigger a rebalance. This approach also reduce memory and disk usage. However, it wastes lot of resources to recalculate same topic hash. For example, if 10 groups subscribe to a same topic, the coordinator need to recalculate it 10 times.

Check Topic Delta to Trigger a Rebalance

Check the changed part like TopicDelta#newPartitons when receiving a new metadata image. If there is change data, set a value in the group and trigger a rebalance in the next group heartbeat. This approach can reduce hash calculation and storage. However, the group coordinator is not always online. If the topic change is not happened with online coordinator, the change will be ignored and the coordinator can't trigger a rebalance.

Add Epoch to TopicImage

A new metadata image is computed by controller. An epoch can represent the version of TopicImage. If there is number of partitions or partition racks change, the controller bumps the epoch. When coordinator receives a new metadata image and there is a new topic epoch, it triggers a rebalance for a group. The group also can store topic epoch map in records, so it can know the difference if the coordinator restarts. This approach save cpu and storage resources and avoid the downside of "Check Topic Delta to Trigger a Rebalance". However, this approach mixes the group coordinator logic within the controller.