dajac commented on code in PR #19523:
URL: https://github.com/apache/kafka/pull/19523#discussion_r2086685337
##########
build.gradle:
##########
@@ -1414,10 +1414,12 @@ project(':group-coordinator') {
implementation project(':coordinator-common')
implementation libs.jacksonDatabind
implementation libs.jacksonJDK8Datatypes
+ implementation libs.lz4
Review Comment:
Do we need this change?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Utils.java:
##########
@@ -324,4 +329,84 @@ 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;
+
+ /**
+ * Computes the hash of the topics in a group.
+ * <p>
+ * The computed hash value is stored as part of the metadata hash in the
*GroupMetadataValue.
+ * <p>
+ * The hashing process involves the following steps:
+ * 1. Sort the topic hashes by topic name.
+ * 2. Write each topic hash in order.
+ *
+ * @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) {
+ // Sort entries by topic name
+ List<Map.Entry<String, Long>> sortedEntries = new
ArrayList<>(topicHashes.entrySet());
+ sortedEntries.sort(Map.Entry.comparingByKey());
+
+ HashStream64 hasher = Hashing.xxh3_64().hashStream();
+ for (Map.Entry<String, Long> entry : sortedEntries) {
+ hasher.putLong(entry.getValue());
+ }
+
+ return hasher.getAsLong();
+ }
+
+ /**
+ * Computes the hash of the topic id, name, number of partitions, and
partition racks by XXHash64.
+ * <p>
+ * The computed hash value for the topic is utilized in conjunction with
the {@link #computeGroupHash(Map)}
+ * method and is stored as part of the metadata hash in the
*GroupMetadataValue.
+ * It is important to note that if the hash algorithm is changed, the
magic byte must be updated to reflect the
+ * new hash version.
+ * <p>
+ * The hashing process involves the following steps:
+ * 1. Write a magic byte to denote the version of the hash function.
+ * 2. Write the hash code of the topic ID.
+ * 3. Write the topic name.
+ * 4. Write the number of partitions associated with the topic.
+ * 5. For each partition, write the partition ID and a sorted list of rack
identifiers.
+ * - Rack identifiers are formatted as
"<length1><value1><length2><value2>" to prevent issues with simple separators.
+ *
+ * @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 {
+ HashStream64 hasher = Hashing.xxh3_64().hashStream();
+ hasher = hasher.putByte(TOPIC_HASH_MAGIC_BYTE) // magic byte
+ .putLong(topicImage.id().hashCode()) // topic ID
+ .putString(topicImage.name()) // topic name
+ .putInt(topicImage.partitions().size()); // number of partitions
+
+ for (int i = 0; i < topicImage.partitions().size(); i++) {
+ hasher = hasher.putInt(i); // partition id
+ // The rack string combination cannot use simple separator like
",", because there is no limitation for rack character.
+ // If using simple separator like "," it may hit edge case like
",," and ",,," / ",,," and ",,".
+ // Add length before the rack string to avoid the edge case.
+ List<String> racks = new ArrayList<>();
Review Comment:
nit: I wonder whether we could reuse this list. We could declare it before
the loop and clear it before using it. What do you think?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Utils.java:
##########
@@ -324,4 +329,84 @@ 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;
+
+ /**
+ * Computes the hash of the topics in a group.
+ * <p>
+ * The computed hash value is stored as part of the metadata hash in the
*GroupMetadataValue.
+ * <p>
+ * The hashing process involves the following steps:
+ * 1. Sort the topic hashes by topic name.
+ * 2. Write each topic hash in order.
+ *
+ * @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) {
+ // Sort entries by topic name
+ List<Map.Entry<String, Long>> sortedEntries = new
ArrayList<>(topicHashes.entrySet());
+ sortedEntries.sort(Map.Entry.comparingByKey());
+
+ HashStream64 hasher = Hashing.xxh3_64().hashStream();
+ for (Map.Entry<String, Long> entry : sortedEntries) {
+ hasher.putLong(entry.getValue());
+ }
+
+ return hasher.getAsLong();
+ }
+
+ /**
+ * Computes the hash of the topic id, name, number of partitions, and
partition racks by XXHash64.
+ * <p>
+ * The computed hash value for the topic is utilized in conjunction with
the {@link #computeGroupHash(Map)}
+ * method and is stored as part of the metadata hash in the
*GroupMetadataValue.
+ * It is important to note that if the hash algorithm is changed, the
magic byte must be updated to reflect the
+ * new hash version.
+ * <p>
+ * The hashing process involves the following steps:
+ * 1. Write a magic byte to denote the version of the hash function.
+ * 2. Write the hash code of the topic ID.
+ * 3. Write the topic name.
+ * 4. Write the number of partitions associated with the topic.
+ * 5. For each partition, write the partition ID and a sorted list of rack
identifiers.
+ * - Rack identifiers are formatted as
"<length1><value1><length2><value2>" to prevent issues with simple separators.
+ *
+ * @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 {
+ HashStream64 hasher = Hashing.xxh3_64().hashStream();
+ hasher = hasher.putByte(TOPIC_HASH_MAGIC_BYTE) // magic byte
Review Comment:
nit: Could you please put `.putByte(TOPIC_HASH_MAGIC_BYTE) // magic byte` on
a new line. It is easier to read if it is aligned with the other ones.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Utils.java:
##########
@@ -324,4 +329,84 @@ 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;
+
+ /**
+ * Computes the hash of the topics in a group.
+ * <p>
+ * The computed hash value is stored as part of the metadata hash in the
*GroupMetadataValue.
+ * <p>
+ * The hashing process involves the following steps:
+ * 1. Sort the topic hashes by topic name.
+ * 2. Write each topic hash in order.
+ *
+ * @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) {
Review Comment:
nit: public?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/UtilsTest.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.image.MetadataImage;
+
+import com.dynatrace.hash4j.hashing.Hashing;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.io.IOException;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+
+public class UtilsTest {
+ private static final Uuid FOO_TOPIC_ID = Uuid.randomUuid();
+ private static final String FOO_TOPIC_NAME = "foo";
+ private static final String BAR_TOPIC_NAME = "bar";
+ private static final int FOO_NUM_PARTITIONS = 2;
+ private static final MetadataImage FOO_METADATA_IMAGE = new
MetadataImageBuilder()
+ .addTopic(FOO_TOPIC_ID, FOO_TOPIC_NAME, FOO_NUM_PARTITIONS)
+ .addRacks()
+ .build();
+
+ @Test
+ void testComputeTopicHash() throws IOException {
+ long result =
Utils.computeTopicHash(FOO_METADATA_IMAGE.topics().getTopic(FOO_TOPIC_ID),
FOO_METADATA_IMAGE.cluster());
Review Comment:
From a usage perspective, I wonder whether we should just pass the topic
name and the metadata image to the method to compute the hash. I suppose that
we will use it like this in the end but I am not 100% sure. What do you think?
##########
build.gradle:
##########
@@ -1414,10 +1414,12 @@ project(':group-coordinator') {
implementation project(':coordinator-common')
implementation libs.jacksonDatabind
implementation libs.jacksonJDK8Datatypes
+ implementation libs.lz4
implementation libs.metrics
implementation libs.hdrHistogram
implementation libs.re2j
implementation libs.slf4jApi
+ implementation libs.hash4j
Review Comment:
I suppose that we need to update the license file too.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Utils.java:
##########
@@ -324,4 +329,84 @@ 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;
+
+ /**
+ * Computes the hash of the topics in a group.
+ * <p>
+ * The computed hash value is stored as part of the metadata hash in the
*GroupMetadataValue.
+ * <p>
+ * The hashing process involves the following steps:
+ * 1. Sort the topic hashes by topic name.
+ * 2. Write each topic hash in order.
+ *
+ * @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) {
+ // Sort entries by topic name
+ List<Map.Entry<String, Long>> sortedEntries = new
ArrayList<>(topicHashes.entrySet());
+ sortedEntries.sort(Map.Entry.comparingByKey());
+
+ HashStream64 hasher = Hashing.xxh3_64().hashStream();
+ for (Map.Entry<String, Long> entry : sortedEntries) {
+ hasher.putLong(entry.getValue());
+ }
+
+ return hasher.getAsLong();
+ }
+
+ /**
+ * Computes the hash of the topic id, name, number of partitions, and
partition racks by XXHash64.
+ * <p>
+ * The computed hash value for the topic is utilized in conjunction with
the {@link #computeGroupHash(Map)}
+ * method and is stored as part of the metadata hash in the
*GroupMetadataValue.
+ * It is important to note that if the hash algorithm is changed, the
magic byte must be updated to reflect the
+ * new hash version.
+ * <p>
+ * The hashing process involves the following steps:
+ * 1. Write a magic byte to denote the version of the hash function.
+ * 2. Write the hash code of the topic ID.
+ * 3. Write the topic name.
+ * 4. Write the number of partitions associated with the topic.
+ * 5. For each partition, write the partition ID and a sorted list of rack
identifiers.
+ * - Rack identifiers are formatted as
"<length1><value1><length2><value2>" to prevent issues with simple separators.
+ *
+ * @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 {
+ HashStream64 hasher = Hashing.xxh3_64().hashStream();
+ hasher = hasher.putByte(TOPIC_HASH_MAGIC_BYTE) // magic byte
+ .putLong(topicImage.id().hashCode()) // topic ID
+ .putString(topicImage.name()) // topic name
+ .putInt(topicImage.partitions().size()); // number of partitions
+
+ for (int i = 0; i < topicImage.partitions().size(); i++) {
+ hasher = hasher.putInt(i); // partition id
+ // The rack string combination cannot use simple separator like
",", because there is no limitation for rack character.
+ // If using simple separator like "," it may hit edge case like
",," and ",,," / ",,," and ",,".
+ // Add length before the rack string to avoid the edge case.
+ List<String> racks = new ArrayList<>();
+ for (int replicaId : topicImage.partitions().get(i).replicas) {
+ BrokerRegistration broker = clusterImage.broker(replicaId);
+ if (broker != null) {
+ Optional<String> rackOptional = broker.rack();
+ rackOptional.ifPresent(racks::add);
Review Comment:
nit: We could combine those two lines into one.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/UtilsTest.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.image.MetadataImage;
+
+import com.dynatrace.hash4j.hashing.Hashing;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.io.IOException;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+
+public class UtilsTest {
+ private static final Uuid FOO_TOPIC_ID = Uuid.randomUuid();
+ private static final String FOO_TOPIC_NAME = "foo";
+ private static final String BAR_TOPIC_NAME = "bar";
+ private static final int FOO_NUM_PARTITIONS = 2;
+ private static final MetadataImage FOO_METADATA_IMAGE = new
MetadataImageBuilder()
+ .addTopic(FOO_TOPIC_ID, FOO_TOPIC_NAME, FOO_NUM_PARTITIONS)
+ .addRacks()
+ .build();
+
+ @Test
+ void testComputeTopicHash() throws IOException {
+ long result =
Utils.computeTopicHash(FOO_METADATA_IMAGE.topics().getTopic(FOO_TOPIC_ID),
FOO_METADATA_IMAGE.cluster());
+
+ long expected = Hashing.xxh3_64().hashStream()
+ .putByte((byte) 0) // magic byte
+ .putLong(FOO_TOPIC_ID.hashCode()) // topic ID
+ .putString(FOO_TOPIC_NAME) // topic name
+ .putInt(FOO_NUM_PARTITIONS) // number of partitions
+ .putInt(0) // partition 0
+ .putInt(5) // length of rack0
+ .putString("rack0") // The first rack in partition 0
+ .putInt(5) // length of rack1
+ .putString("rack1") // The second rack in partition 0
+ .putInt(1) // partition 1
+ .putInt(5) // length of rack0
+ .putString("rack1") // The first rack in partition 1
+ .putInt(5) // length of rack1
+ .putString("rack2") // The second rack in partition 1
+ .getAsLong();
+ assertEquals(expected, result);
+ }
+
+ @Test
+ void testComputeTopicHashWithDifferentMagicByte() throws IOException {
+ long result =
Utils.computeTopicHash(FOO_METADATA_IMAGE.topics().getTopic(FOO_TOPIC_ID),
FOO_METADATA_IMAGE.cluster());
+
+ long expected = Hashing.xxh3_64().hashStream()
+ .putByte((byte) 1) // magic byte
+ .putLong(FOO_TOPIC_ID.hashCode()) // topic ID
+ .putString(FOO_TOPIC_NAME) // topic name
+ .putInt(FOO_NUM_PARTITIONS) // number of partitions
+ .putInt(0) // partition 0
+ .putInt(5) // length of rack0
+ .putString("rack0") // The first rack in partition 0
+ .putInt(5) // length of rack1
+ .putString("rack1") // The second rack in partition 0
+ .putInt(1) // partition 1
+ .putInt(5) // length of rack0
+ .putString("rack1") // The first rack in partition 1
+ .putInt(5) // length of rack1
+ .putString("rack2") // The second rack in partition 1
+ .getAsLong();
+ assertNotEquals(expected, result);
+ }
+
+ @Test
+ void testComputeTopicHashWithDifferentPartitionOrder() throws IOException {
+ long result =
Utils.computeTopicHash(FOO_METADATA_IMAGE.topics().getTopic(FOO_TOPIC_ID),
FOO_METADATA_IMAGE.cluster());
+
+ long expected = Hashing.xxh3_64().hashStream()
+ .putByte((byte) 1) // magic byte
+ .putLong(FOO_TOPIC_ID.hashCode()) // topic ID
+ .putString(FOO_TOPIC_NAME) // topic name
+ .putInt(FOO_NUM_PARTITIONS) // number of partitions
+ .putInt(1) // partition 1
+ .putInt(5) // length of rack0
+ .putString("rack1") // The first rack in partition 1
+ .putInt(5) // length of rack1
+ .putString("rack2") // The second rack in partition 1
+ .putInt(0) // partition 0
+ .putInt(5) // length of rack0
+ .putString("rack0") // The first rack in partition 0
+ .putInt(5) // length of rack1
+ .putString("rack1") // The second rack in partition 0
+ .getAsLong();
+ assertNotEquals(expected, result);
+ }
+
+ @Test
+ void testComputeTopicHashWithDifferentRackOrder() throws IOException {
+ long result =
Utils.computeTopicHash(FOO_METADATA_IMAGE.topics().getTopic(FOO_TOPIC_ID),
FOO_METADATA_IMAGE.cluster());
+
+ long expected = Hashing.xxh3_64().hashStream()
+ .putByte((byte) 0) // magic byte
+ .putLong(FOO_TOPIC_ID.hashCode()) // topic ID
+ .putString(FOO_TOPIC_NAME) // topic name
+ .putInt(FOO_NUM_PARTITIONS) // number of partitions
+ .putInt(0) // partition 0
+ .putInt(5) // length of rack0
+ .putString("rack1") // The second rack in partition 0
+ .putInt(5) // length of rack1
+ .putString("rack0") // The first rack in partition 0
+ .putInt(1) // partition 1
+ .putInt(5) // length of rack0
+ .putString("rack1") // The first rack in partition 1
+ .putInt(5) // length of rack1
+ .putString("rack2") // The second rack in partition 1
+ .getAsLong();
+ assertNotEquals(expected, result);
+ }
+
+ @ParameterizedTest
+ @MethodSource("differentFieldGenerator")
+ void testComputeTopicHashWithDifferentField(MetadataImage differentImage,
Uuid topicId) throws IOException {
+ long result =
Utils.computeTopicHash(FOO_METADATA_IMAGE.topics().getTopic(FOO_TOPIC_ID),
FOO_METADATA_IMAGE.cluster());
+
+ assertNotEquals(
+ Utils.computeTopicHash(
+ differentImage.topics().getTopic(topicId),
+ differentImage.cluster()
+ ),
+ result
+ );
+ }
+
+ private static Stream<Arguments> differentFieldGenerator() {
+ Uuid differentTopicId = Uuid.randomUuid();
+ return Stream.of(
+ Arguments.of(new MetadataImageBuilder() // different topic id
+ .addTopic(differentTopicId, FOO_TOPIC_NAME,
FOO_NUM_PARTITIONS)
+ .addRacks()
+ .build(),
+ differentTopicId
+ ),
Review Comment:
nit: Could you please put `new MetadataImageBuilder() // different topic id`
on a new line too?
```
Arguments.of(
new MetadataImageBuilder() // different topic id
.addTopic(differentTopicId, FOO_TOPIC_NAME,
FOO_NUM_PARTITIONS)
.addRacks()
.build(),
differentTopicId
),
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Utils.java:
##########
@@ -324,4 +329,84 @@ 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;
+
+ /**
+ * Computes the hash of the topics in a group.
+ * <p>
+ * The computed hash value is stored as part of the metadata hash in the
*GroupMetadataValue.
+ * <p>
+ * The hashing process involves the following steps:
+ * 1. Sort the topic hashes by topic name.
+ * 2. Write each topic hash in order.
+ *
+ * @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) {
+ // Sort entries by topic name
+ List<Map.Entry<String, Long>> sortedEntries = new
ArrayList<>(topicHashes.entrySet());
+ sortedEntries.sort(Map.Entry.comparingByKey());
+
+ HashStream64 hasher = Hashing.xxh3_64().hashStream();
+ for (Map.Entry<String, Long> entry : sortedEntries) {
+ hasher.putLong(entry.getValue());
+ }
+
+ return hasher.getAsLong();
+ }
+
+ /**
+ * Computes the hash of the topic id, name, number of partitions, and
partition racks by XXHash64.
+ * <p>
+ * The computed hash value for the topic is utilized in conjunction with
the {@link #computeGroupHash(Map)}
+ * method and is stored as part of the metadata hash in the
*GroupMetadataValue.
+ * It is important to note that if the hash algorithm is changed, the
magic byte must be updated to reflect the
+ * new hash version.
+ * <p>
+ * The hashing process involves the following steps:
+ * 1. Write a magic byte to denote the version of the hash function.
+ * 2. Write the hash code of the topic ID.
+ * 3. Write the topic name.
+ * 4. Write the number of partitions associated with the topic.
+ * 5. For each partition, write the partition ID and a sorted list of rack
identifiers.
+ * - Rack identifiers are formatted as
"<length1><value1><length2><value2>" to prevent issues with simple separators.
+ *
+ * @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 {
+ HashStream64 hasher = Hashing.xxh3_64().hashStream();
+ hasher = hasher.putByte(TOPIC_HASH_MAGIC_BYTE) // magic byte
+ .putLong(topicImage.id().hashCode()) // topic ID
+ .putString(topicImage.name()) // topic name
+ .putInt(topicImage.partitions().size()); // number of partitions
+
+ for (int i = 0; i < topicImage.partitions().size(); i++) {
+ hasher = hasher.putInt(i); // partition id
+ // The rack string combination cannot use simple separator like
",", because there is no limitation for rack character.
+ // If using simple separator like "," it may hit edge case like
",," and ",,," / ",,," and ",,".
+ // Add length before the rack string to avoid the edge case.
+ List<String> racks = new ArrayList<>();
+ for (int replicaId : topicImage.partitions().get(i).replicas) {
+ BrokerRegistration broker = clusterImage.broker(replicaId);
+ if (broker != null) {
+ Optional<String> rackOptional = broker.rack();
+ rackOptional.ifPresent(racks::add);
+ }
+ }
+
+ Collections.sort(racks);
+ for (String rack : racks) {
+ // Format: "<length><value>"
+ hasher = hasher.putInt(rack.length()).putString(rack);
+ }
+ }
+ return hasher.getAsLong();
Review Comment:
nit: Should we put an empty line before the return?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Utils.java:
##########
@@ -324,4 +329,84 @@ 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;
+
+ /**
+ * Computes the hash of the topics in a group.
+ * <p>
+ * The computed hash value is stored as part of the metadata hash in the
*GroupMetadataValue.
+ * <p>
+ * The hashing process involves the following steps:
+ * 1. Sort the topic hashes by topic name.
+ * 2. Write each topic hash in order.
+ *
+ * @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) {
+ // Sort entries by topic name
+ List<Map.Entry<String, Long>> sortedEntries = new
ArrayList<>(topicHashes.entrySet());
+ sortedEntries.sort(Map.Entry.comparingByKey());
+
+ HashStream64 hasher = Hashing.xxh3_64().hashStream();
+ for (Map.Entry<String, Long> entry : sortedEntries) {
+ hasher.putLong(entry.getValue());
+ }
+
+ return hasher.getAsLong();
+ }
+
+ /**
+ * Computes the hash of the topic id, name, number of partitions, and
partition racks by XXHash64.
+ * <p>
+ * The computed hash value for the topic is utilized in conjunction with
the {@link #computeGroupHash(Map)}
+ * method and is stored as part of the metadata hash in the
*GroupMetadataValue.
+ * It is important to note that if the hash algorithm is changed, the
magic byte must be updated to reflect the
+ * new hash version.
+ * <p>
+ * The hashing process involves the following steps:
+ * 1. Write a magic byte to denote the version of the hash function.
+ * 2. Write the hash code of the topic ID.
+ * 3. Write the topic name.
+ * 4. Write the number of partitions associated with the topic.
+ * 5. For each partition, write the partition ID and a sorted list of rack
identifiers.
+ * - Rack identifiers are formatted as
"<length1><value1><length2><value2>" to prevent issues with simple separators.
+ *
+ * @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:
nit: public?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Utils.java:
##########
@@ -324,4 +329,84 @@ 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;
+
+ /**
+ * Computes the hash of the topics in a group.
+ * <p>
+ * The computed hash value is stored as part of the metadata hash in the
*GroupMetadataValue.
+ * <p>
+ * The hashing process involves the following steps:
+ * 1. Sort the topic hashes by topic name.
+ * 2. Write each topic hash in order.
+ *
+ * @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) {
+ // Sort entries by topic name
+ List<Map.Entry<String, Long>> sortedEntries = new
ArrayList<>(topicHashes.entrySet());
+ sortedEntries.sort(Map.Entry.comparingByKey());
+
+ HashStream64 hasher = Hashing.xxh3_64().hashStream();
+ for (Map.Entry<String, Long> entry : sortedEntries) {
+ hasher.putLong(entry.getValue());
+ }
+
+ return hasher.getAsLong();
+ }
+
+ /**
+ * Computes the hash of the topic id, name, number of partitions, and
partition racks by XXHash64.
+ * <p>
+ * The computed hash value for the topic is utilized in conjunction with
the {@link #computeGroupHash(Map)}
+ * method and is stored as part of the metadata hash in the
*GroupMetadataValue.
+ * It is important to note that if the hash algorithm is changed, the
magic byte must be updated to reflect the
+ * new hash version.
+ * <p>
+ * The hashing process involves the following steps:
+ * 1. Write a magic byte to denote the version of the hash function.
+ * 2. Write the hash code of the topic ID.
+ * 3. Write the topic name.
+ * 4. Write the number of partitions associated with the topic.
+ * 5. For each partition, write the partition ID and a sorted list of rack
identifiers.
+ * - Rack identifiers are formatted as
"<length1><value1><length2><value2>" to prevent issues with simple separators.
+ *
+ * @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 {
+ HashStream64 hasher = Hashing.xxh3_64().hashStream();
+ hasher = hasher.putByte(TOPIC_HASH_MAGIC_BYTE) // magic byte
+ .putLong(topicImage.id().hashCode()) // topic ID
+ .putString(topicImage.name()) // topic name
+ .putInt(topicImage.partitions().size()); // number of partitions
+
+ for (int i = 0; i < topicImage.partitions().size(); i++) {
+ hasher = hasher.putInt(i); // partition id
+ // The rack string combination cannot use simple separator like
",", because there is no limitation for rack character.
+ // If using simple separator like "," it may hit edge case like
",," and ",,," / ",,," and ",,".
+ // Add length before the rack string to avoid the edge case.
Review Comment:
Should we rather put this after `// Format: "<length><value>"`? It seems
more related to that part of the code.
--
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
