DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
A single hash of all subscribed topic in Group
The group doesn't need to store subscribed topic metadata, because the coordinator doesn't need the value to detect a rebalance. After In this KIP, the group gets all subscribed topic hash from the cache in coordinator and sums them as a single hash. The final hash will be stored with a new group epoch in ConsumerGroupMetadataValue / ShareGroupMetadataValue / StreamGroupMetadataValue.
Topic Hash Function
A topic hash represents topic id, name, number of partitions, and partition racks. To avoid useless rebalance, the hash function should return same value for same data, even if it runs on different JDKs or partition racks have different order. For different JDKs, the KIP will use Murmur3 to compute the hash value. For partition racks with different order, we will compute hash for each value in it and sum as a result. We will also set the first byte as magic byte to represent hash version.
| Code Block | ||||
|---|---|---|---|---|
| ||||
import com.google.common.hash.HashCode;
import com.google.common.hash.HashFunction;
import com.google.common.hash.Hasher;
import com.google.common.hash.Hashing;
public abstract class ModernGroup<T extends ModernGroupMember> implements Group {
// ...
/**
* Compute metadata hash based on the current subscription info.
*
* @param subscribedTopicNames Map of topic names to the number of subscribers.
* @param metadataImage The current metadata for all available topics.
* @param topicHashCache The cache of topic hashes.
*/
public long computeMetadataHash(
Map<String, SubscriptionCount> subscribedTopicNames,
MetadataImage metadataImage,
Map<String, Long> topicHashCache
) {
TopicsImage topicsImage = metadataImage.topics();
List<HashCode> hashCodes = subscribedTopicNames.keySet().stream()
.filter(topicName -> topicsImage.getTopic(topicName) != null)
.map(topicName -> HashCode.fromLong(
topicHashCache.computeIfAbsent(
topicName,
key -> computeTopicHash(topicsImage.getTopic(topicName), metadataImage.cluster())
)
))
.toList();
return hashCodes.isEmpty() ? 0 : Hashing.combineUnordered(
hashCodes
).asLong();
}
} |
Topic Hash Function
A topic hash represents topic id, name, number of partitions, and partition racks. To avoid useless rebalance, the hash function should return same value for same data, even if it runs on different JDKs or partition racks have different order. For different JDKs, the KIP will use Murmur3 to compute the hash value. For partition racks with different order, we will compute hash for each value in it and sum as a result. We will also set the first byte as magic byte to represent hash version.
| Code Block | ||
|---|---|---|
| ||
import com.google.common.hash.HashCode;
import com.google.common.hash.HashFunction;
import com.google.common.hash.Hasher;
import com.google.common.hash.Hashing;
public abstract class ModernGroup<T extends ModernGroupMember> implements Group {
// ...
/**
* 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 static long computeTopicHash(TopicImage topicImage, ClusterImage clusterImage) {
HashFunction hf = Hashing.murmur3_128();
Hasher topicHasher = hf.newHasher()
.putByte((byte) 0) // magic byte
.putLong(topicImage.id().hashCode()) // topic Id
.putString(topicImage.name(), StandardCharsets.UTF_8) // topic name
.putLong(topicImage.partitions().size()); // number of partitions
topicImage.partitions().entrySet().stream().sorted(Map.Entry.comparingByKey()).forEach(entry -> {
topicHasher.putInt(entry.getKey()); // partition id
Arrays.stream(entry.getValue().replicas)
.mapToObj(clusterImage::broker)
.filter(Objects::nonNull)
.map(BrokerRegistration::rack)
.filter(Optional::isPresent)
.map(Optional::get)
.sorted()
.forEach(rack -> topicHasher.putString(rack, StandardCharsets.UTF_8)); // sorted racks
});
return topicHasher.hash().asLong();
}
} | ||
| Code Block | ||
| ||
public static long topicHash(String topicName, int numPartitions, Map<Integer, List<Integer>> partitionIdToSortedRacks) |
SubscribedTopicDescriberImpl
...