Current state: Under Discussion
Discussion thread: https://lists.apache.org/thread/l8ko353v3nn1blgymsty895x6c98oxlx
JIRA:
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
KIP-848 introduces a next generation of consumer rebalance protocol which supports rack-aware partition assignment. In the first implement, the group coordinator computes subscribed topic metadata which contains topic UUID, name, number of partition, and rack set of each partition. If the metadata is expired, the group coordinator computes new subscribed topic metadata and compare with current one. If result is difference, the group coordinator bumps group epoch and calculates new target assignment which means it triggers a rebalance. However, the rack set of each partition takes too much space. In KAFKA-17578, a real case of memory usage shows that in a group with 500 members and 2K topic partitions, partition racks account for 79% of whole ConsumerGroup object. This KIP will get rid of subscribed topic metadata and use a subscribed topic hash map to replace it. Each topic has a hash value. There are two server side conditions can trigger a new rebalance, so the topic hash needs to reflect them:
Add a new field TopicHash.
{
"type": "data",
"name": "ConsumerGroupPartitionMetadataValue",
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "Topics", "versions": "0+", "type": "[]TopicMetadata",
"about": "The list of topic metadata.", "fields": [
{ "name": "TopicId", "versions": "0+", "type": "uuid",
"about": "The topic id." },
{ "name": "TopicName", "versions": "0+", "type": "string",
"about": "The topic name." },
{ "name": "NumPartitions", "versions": "0+", "type": "int32",
"about": "The number of partitions of the topic." },
{ "name": "PartitionMetadata", "versions": "0+", "type": "[]PartitionMetadata",
"about": "Deprecated: this field is not used after 4.0. Partitions mapped to a set of racks. If the rack information is unavailable for all the partitions, an empty list is stored", "fields": [
{ "name": "Partition", "versions": "0+", "type": "int32",
"about": "The partition number." },
{ "name": "Racks", "versions": "0+", "type": "[]string",
"about": "The set of racks that the partition is mapped to." }
]},
{ "name": "TopicHash", "versions": "0+", "type": "int64",
"default": 0, "taggedVersions": "0+", "tag": 0,
"about": "The hash value of the topic id, name, number of partitions, and partition racks." }
]}
]
} |
Add a new field TopicHash.
{
"type": "data",
"name": "ShareGroupPartitionMetadataValue",
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "Topics", "versions": "0+", "type": "[]TopicMetadata",
"about": "The list of topic metadata.", "fields": [
{ "name": "TopicId", "versions": "0+", "type": "uuid",
"about": "The topic id." },
{ "name": "TopicName", "versions": "0+", "type": "string",
"about": "The topic name." },
{ "name": "NumPartitions", "versions": "0+", "type": "int32",
"about": "The number of partitions of the topic." },
{ "name": "PartitionMetadata", "versions": "0+", "type": "[]PartitionMetadata",
"about": "Partitions mapped to a set of racks. If the rack information is unavailable for all the partitions, an empty list is stored", "fields": [
{ "name": "Partition", "versions": "0+", "type": "int32",
"about": "The partition number." },
{ "name": "Racks", "versions": "0+", "type": "[]string",
"about": "The set of racks that the partition is mapped to." }
]},
{ "name": "TopicHash", "versions": "0+", "type": "int64",
"default": 0, "taggedVersions": "0+", "tag": 0,
"about": "The hash value of the topic id, name, number of partitions, and partition racks." }
]}
]
} |
Different groups may subscribe to same topics. With topic hash map cache in the coordinator, it can avoid recalculation of same topic hash for different groups. The topic hash value should represent topic id, name, number of partitions, and partition racks.
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 replaying ConsumerGroupMemberMetadtaValue / ShareGroupMemberMetadataValue, 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.
A topic will be removed from groupsByTopics if there is no group subscribe to it. The group coordinator can remove the topic hash as well, so it can leverage same mechanism to remove a topic hash.
When there is a new metadata image, it contains changed topics and deleted topics in metadata delta. For both cases, the coordinator remove topic hash from the cache. The value will be recomputed when in the next consumer group heartbeat. The lazy evaluation is more efficient. If there is no further group heartbeat, the new topic hash value is useless, so we can just compute it in group heartbeat.
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);
allGroupIds.addAll(groupsSubscribedToTopic(topicName));
});
topicsDelta.deletedTopicIds().forEach(topicId -> {
TopicImage topicImage = delta.image().topics().getTopic(topicId);
String topicName = topicImage.name();
topicHashCache.remove(topicName);
allGroupIds.addAll(groupsSubscribedToTopic(topicName));
});
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. If a topic hash is calculated by another group, we can just reuse it from cache.
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,
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)));
}
});
return Collections.unmodifiableMap(newSubscriptionTopicHash);
}
} |
Since the group coordinator doesn't compute subscribed topic metadata, the SubscribedTopicDescribeImpl can't use Map<Uuid, TopicMetadata> to number of partition and rack set. We will use MetadataImage as input to replace it.
There is new field in ConsumerGroupPartitionMetadataValue and ShareGroupPartitionMetadataValue. For compatibility, adding it as tagged field, so old brokers can deserialize it.
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 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.
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.