DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
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 When 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 number of partition and rack set of each partition. It uses 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:
- A topic with a new partition.
- A topic partition has rack change. Each topic partition has multiple replicas. Each replica is stored on a broker. The rack value is from the broker.rack config. It's a read only config and can only be changed when broker restarts. If a broker stops, a related replica will be removed from topic partitions. If a broker starts, it will be added to topic partitions.
Following table compares cpu / memory usage of different strategies:
| CPU | Memory | |
|---|---|---|
1. Subscription topic metadata with topic UUID, name, number of partition, and rack set of each partition. | Low | High |
| 2. Subscription topic metadata with topic UUID, name, and hash. | Mid | Mid |
| 3. A single hash to represent all subscribed topic in a group. | High | Low |
We choose the second way, but not the third way, because a single hash to represent all subscribed topic in a group wastes too much CPU resource to recalculate the hash when the metadata image is expired. If any subscribed topic change, the third way needs to aggregate all data an recalculate the hash. The second way only needs to recalculate expired topic hashCompare the subscribed topic metadata, a single hash of the subscribed topic metadata, and the subscribed topic hash map. The subscribed topic metadata requires the most storage but the least computation. A single hash of the subscribed topic metadata requires the least storage but the most computation. A balanced approach would aim to optimize both computation and storage usage. This KIP uses a hash value for each topic, so a new hash only needs to be computed when there is a change to the topic. Each topic only stores a hash value, so we don’t need to store all detailed information.
Public Interfaces
ConsumerGroupPartitionMetadataValue
Add a new tagged field TopicHash.
| Code Block | ||||
|---|---|---|---|---|
| ||||
{
"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": "Deprecated: this field is not used after 4.0. The number of partitions of the topic." }, <-- deprecated field
{ "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." }
]}, <-- deprecated field
{ "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." } <-- new field
]}
]
} |
ShareGroupPartitionMetadataValue
Add a new tagged field TopicHash.
| Code Block | ||||
|---|---|---|---|---|
| ||||
{
"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": "Deprecated: this field is not used after 4.0. The number of partitions of the topic." },, <-- deprecated field
{ "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." }
]}, <-- deprecated field
{ "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." } <-- new field
]}
]
} |
Proposed Changes
Topic Hash Map Cache in Coordinator
...
| Code Block | ||||
|---|---|---|---|---|
| ||||
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;
// Initialize the last offset if it was not yet.
if (lastMetadataImageWithNewTopics == -1L) {
lastMetadataImageWithNewTopics = metadataImage.provenance().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);
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();
}
});
}
} |
Subscribed Topic Hash Map in Group
...
Map in Group
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.
...
| Code Block | ||||
|---|---|---|---|---|
| ||||
public abstract class ModernGroup<T extends ModernGroupMember> implements Group {
// ...
/**
* TheComputes hashthe withsubscription eachmetadata subscribedbased topicon name.
the current */subscription info.
protected final TimelineHashMap<String, Long> subscriptionTopicHash;
*
/ **
@param subscribedTopicNames Map of *topic @returnnames Anto immutablethe Mapnumber of subscription topic hash forsubscribers.
* @param metadataImage each topicThe thatcurrent themetadata consumerfor groupall isavailable subscribed totopics.
*/
@param topicHashCache public Map<String, Long> subscriptionTopicHash() {
The cache of return Collections.unmodifiableMap(subscriptionTopicHash);topic hashes.
* }
@return An immutable map /**
of subscription topic * Updates the subscriptionhash for each topic that hash.the Thisconsumer replacesgroup theis previoussubscribed oneto.
*/
public Map<String, TopicMetadata> computeSubscriptionMetadata(
* @param subscriptionTopicHash The new subscription topic hash.
Map<String, SubscriptionCount> subscribedTopicNames,
*/
public voidMetadataImage setSubscriptionTopicHash(metadataImage,
Map<String, Long> subscriptionTopicHashtopicHashCache
) {
// Create the topic metadata for each subscribed topic.
this.subscriptionTopicHash.clear();
Map<String, TopicMetadata> newSubscriptionMetadata = new this.subscriptionTopicHash.putAll(subscriptionTopicHashHashMap<>(subscribedTopicNames.size());
}
/**
TopicsImage topicsImage = metadataImage.topics();
* Computes the subscription topic hash based on the current subscription info.
*
subscribedTopicNames.forEach((topicName, count) -> {
* @param subscribedTopicNames Map of topic names toTopicImage thetopicImage number= of subscribers.topicsImage.getTopic(topicName);
* @param topicHashCache if (topicImage != The current topic hash cache from coordinator.null) {
*
* @return An immutable map of subscription topic hash for each topic that the consumer group is subscribed to.
newSubscriptionMetadata.put(topicName, new TopicMetadata(
*/
public Map<String, Long> computeSubscriptionTopicHashtopicImage.id(),
Map<String, Integer> subscribedTopicNames,
TopicsImage topicsImagetopicImage.name(),
ClusterImage clusterImage,
Map<String, Long> topicHashCache.computeIfAbsent(
) {
// Create the topic hash for each subscribed topic.
topicName,
Map<String, Long> newSubscriptionTopicHash = new HashMap<>();
subscribedTopicNames.forEach((topicName, count) -> {
key TopicImage-> computeTopicHash(topicImage, = topicsImagemetadataImage.getTopiccluster(topicName));
if (topicImage != null) {
newSubscriptionTopicHash.put(topicName, topicHashCache.computeIfAbsent(topicName, t -> computeTopicHash(topicImage, clusterImage))) )
));
}
});
return Collections.unmodifiableMap(newSubscriptionTopicHashnewSubscriptionMetadata);
}
/**
* Computes the hash of the topic id, name, number of partitions, and partition racks by Murmur3.
*
* @param topicImage The topic image.
* @param clusterImage The cluster image.
*/
public privatestatic long computeTopicHash(TopicImage topicImage, ClusterImage clusterImage) {
long result = topicImage.id().hashCode();
result = 63 * result + Murmur3.hash64(topicImage.name().getBytes(StandardCharsets.UTF_8));
result = 63 * result + Murmur3.hash64(topicImage.partitions().size());
result = 63 * result + topicImage.partitions().entrySet().stream().mapToLong(
entry -> {
PartitionRegistration partitionRegistration = entry.getValue();
long partitionHash = Murmur3.hash64(entry.getKey());
Set<String> racks = Arrays.stream(partitionRegistration.replicas)
.mapToObj(clusterImage::broker)
.map(BrokerRegistration::rack)
.filter(Optional::isPresent)
.map(Optional::get)
.collect(Collectors.toSet());
return 63 * partitionHash + racks.stream().mapToLong(
rack -> Murmur3.hash64(rack.getBytes(StandardCharsets.UTF_8))
).reduce(0L, Long::sum);
}
).reduce(0L, Long::sum);
return result;
}
} |
SubscribedTopicDescribeImpl
SubscribedTopicDescriberImpl
The topic metadata Since the group coordinator doesn't compute subscribed topic metadata, the SubscribedTopicDescribeImpl can't have number of partitions and rack data, so the subscribedTopicDescriberImpl can't rely on it to return numPartitions and racksForPartition. This KIP adds MetadataImage to SubscribedTopicDescriberImpl constructor, so it can use Map<Uuid, TopicMetadata> to number of partition and rack set. We will use MetadataImage as input to replace itcheck which topic is subscribed and use MetadataImage to return numPartitions and racksForPartition.
Compatibility, Deprecation, and Migration Plan
...