chia7712 commented on code in PR #19523:
URL: https://github.com/apache/kafka/pull/19523#discussion_r2800928851
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Group.java:
##########
@@ -209,4 +219,50 @@ void validateOffsetFetch(
default boolean shouldExpire() {
return true;
}
+
+ /**
+ * Computes the hash of the topics in a group.
+ *
+ * @param topicHashes The map of topic hashes. Key is topic name and value
is the topic hash.
+ * @return The hash of the group.
+ */
+ static long computeGroupHash(Map<String, Long> topicHashes) {
+ return Hashing.combineOrdered(
+ topicHashes.entrySet()
+ .stream()
+ .sorted(Map.Entry.comparingByKey())
+ .map(e -> HashCode.fromLong(e.getValue()))
+ .toList()
+ ).asLong();
+ }
+
+ /**
+ * 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.
+ * @return The hash of the topic.
+ */
+ 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
+ .putInt(topicImage.partitions().size()); // number of partitions
+
+
topicImage.partitions().entrySet().stream().sorted(Map.Entry.comparingByKey()).forEach(entry
-> {
+ topicHasher.putInt(entry.getKey()); // partition id
+ String racks = Arrays.stream(entry.getValue().replicas)
+ .mapToObj(clusterImage::broker)
+ .filter(Objects::nonNull)
+ .map(BrokerRegistration::rack)
+ .filter(Optional::isPresent)
+ .map(Optional::get)
+ .sorted()
+ .collect(Collectors.joining(";"));
Review Comment:
Looking at the source code, `putString` already handles he collision issue
by appending the length:
```java
@Override
default HashStream putString(String s) {
putChars(s);
putInt(s.length());
return this;
}
```
https://github.com/dynatrace-oss/hash4j/blob/main/src/main/java/com/dynatrace/hash4j/hashing/AbstractHashStream.java
The cost of this extra length info is negligible, so we don't need to change
the logic. However, it would be helpful to amend the comments to clarify that.
WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
