chia7712 commented on code in PR #19523:
URL: https://github.com/apache/kafka/pull/19523#discussion_r2068885949
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Utils.java:
##########
@@ -324,4 +337,87 @@ static void throwIfRegularExpressionIsInvalid(
regex, ex.getDescription()));
}
}
+
+ /**
+ * The magic byte used to identify the version of topic hash function.
+ */
+ static final byte TOPIC_HASH_MAGIC_BYTE = 0x00;
+ static final XXHash64 LZ4_HASH_INSTANCE =
XXHashFactory.fastestInstance().hash64();
+
+ /**
+ * 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) {
+ // Convert long to byte array. This is taken from guava
LongHashCode#asBytes.
+ //
https://github.com/google/guava/blob/bdf2a9d05342fca852645278d474082905e09d94/guava/src/com/google/common/hash/HashCode.java#L187-L199
+ LongFunction<byte[]> longToBytes = (long value) -> new byte[] {
+ (byte) value,
+ (byte) (value >> 8),
+ (byte) (value >> 16),
+ (byte) (value >> 24),
+ (byte) (value >> 32),
+ (byte) (value >> 40),
+ (byte) (value >> 48),
+ (byte) (value >> 56)
+ };
+
+ // Combine the sorted topic hashes.
+ byte[] resultBytes = new byte[8];
+ topicHashes.entrySet()
+ .stream()
+ .sorted(Map.Entry.comparingByKey()) // sort by topic name
+ .map(Map.Entry::getValue)
+ .map(longToBytes::apply)
+ .forEach(nextBytes -> {
+ // Combine ordered hashes. This is taken from guava
Hashing#combineOrdered.
+ //
https://github.com/google/guava/blob/bdf2a9d05342fca852645278d474082905e09d94/guava/src/com/google/common/hash/Hashing.java#L689-L712
+ for (int i = 0; i < nextBytes.length; i++) {
+ resultBytes[i] = (byte) (resultBytes[i] * 37 ^
nextBytes[i]);
+ }
+ });
+
+ return LZ4_HASH_INSTANCE.hash(resultBytes, 0, resultBytes.length, 0);
+ }
+
+ /**
+ * Computes the hash of the topic id, name, number of partitions, and
partition racks by XXHash64.
+ *
+ * @param topicImage The topic image.
+ * @param clusterImage The cluster image.
+ * @return The hash of the topic.
+ */
+ static long computeTopicHash(TopicImage topicImage, ClusterImage
clusterImage) throws IOException {
Review Comment:
Please add documentation to remind developers that the hash is stored as
part of the state. Changing the implementation of the hashing function may
break compatibility with existing states.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
