FrankYang0529 commented on code in PR #19523:
URL: https://github.com/apache/kafka/pull/19523#discussion_r2068862253
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Group.java:
##########
@@ -209,4 +221,90 @@ void validateOffsetFetch(
default boolean shouldExpire() {
return true;
}
+
+ /**
+ * The magic byte used to identify the version of topic hash function.
+ */
+ byte TOPIC_HASH_MAGIC_BYTE = 0x00;
+ 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]);
+ }
+ });
+
+ // Convert the byte array to long. This is taken from guava
BytesHashCode#asLong.
+ //
https://github.com/google/guava/blob/bdf2a9d05342fca852645278d474082905e09d94/guava/src/com/google/common/hash/HashCode.java#L279-L295
+ long retVal = (resultBytes[0] & 0xFF);
+ for (int i = 1; i < resultBytes.length; i++) {
+ retVal |= (resultBytes[i] & 0xFFL) << (i * 8);
+ }
+ return retVal;
+ }
+
+ /**
+ * 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) throws IOException {
+ try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
Review Comment:
Sorry, I misunderstood ByteBufferOutputStream. I thought it uses fixed
capacity even if there is no enough buffer. After checking the source code, it
expands memory if the buffer is not big enough. Updated it. Thanks.
--
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
