KAFKA-17747: Add compute topic and group hash #51
Conversation
Signed-off-by: PoAn Yang <[email protected]>
|
CodeAnt AI is reviewing your PR. Thanks for using CodeAnt! 🎉We're free for open-source projects. if you're enjoying it, help us grow by sharing. Share on X · |
Summary of ChangesHello @visz11, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request enhances the Highlights
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
|
Note Other AI code review bot(s) detectedCodeRabbit has detected other AI code review bot(s) in this pull request and will avoid duplicating their findings in the review comments. This may lead to a less comprehensive review. WalkthroughThe PR introduces the Guava hashing library dependency to the group-coordinator module and implements two static utility methods for computing deterministic hashes of topic metadata and group topic maps using Murmur3-128 hashing, with supporting configuration and comprehensive test coverage. Changes
Sequence DiagramsequenceDiagram
participant Code as Client Code
participant Group as Group (Interface)
participant Guava as Guava Hashing
participant Cluster as ClusterImage
Note over Code,Group: Computing Topic Hash
Code->>Group: computeTopicHash(topicImage, clusterImage)
Group->>Guava: Hashing.murmur3_128()
Group->>Group: Add magic byte, topic id, name, partitions
Group->>Cluster: Get broker registrations for each partition
Group->>Group: Collect & sort racks per partition
Group->>Guava: putString() with rack order
Group->>Guava: hash()
Guava-->>Group: HashCode result
Group-->>Code: long (hash value)
Note over Code,Group: Computing Group Hash
Code->>Group: computeGroupHash(topicHashes)
Group->>Guava: Hashing.combineOrdered()
Group->>Group: Sort topic entries by key
Group->>Guava: Add all topic hash values
Guava-->>Group: Combined HashCode
Group-->>Code: long (combined hash)
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes
Poem
Pre-merge checks and finishing touches❌ Failed checks (2 warnings)
✅ Passed checks (1 passed)
✨ Finishing touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
@refacto-visz |
codeant-ai
bot
added
the
size:L
feat: KAFKA-17747 Add methods to compute topic and group hashesTL;DR: This PR introduces new utility methods in the Refacto PR SummaryThis change adds the Guava library as a dependency to the The Change HighlightsClick to expand
Sequence DiagramsequenceDiagram
participant Caller as Coordinator/Service
participant Group as Group (static methods)
participant TopicImage as TopicImage
participant ClusterImage as ClusterImage
Caller->>Caller: For each topic in group (e.g., "foo", "bar")
Caller->>Group: computeTopicHash(topicImage, clusterImage)
Group->>TopicImage: Get ID, name, partitions
Group->>ClusterImage: Get broker rack for replicas
Group->>Group: Hash(magic_byte, id, name, num_partitions, sorted_partitions_with_sorted_racks)
Group-->>Caller: topicHash (long)
Caller->>Caller: Collect all topic hashes into Map<TopicName, TopicHash>
Caller->>Group: computeGroupHash(topicHashesMap)
Group->>Group: Sort topic hashes by topic name
Group->>Group: Combine ordered hashes
Group-->>Caller: groupHash (long)
Testing GuideClick to expandThis change introduces utility methods and is primarily validated through unit tests.
|
|
CodeAnt AI finished reviewing your PR. |
|
Refacto is reviewing this PR. Please wait for the review comments to be posted. |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces hashing capabilities for topics and consumer groups by adding the Guava library. The new methods computeTopicHash and computeGroupHash are well-structured and include deterministic sorting, which is great. My review includes a critical suggestion to improve the correctness of UUID hashing by using the full 128 bits instead of hashCode(), and a couple of suggestions to improve code maintainability and readability in both the implementation and the new tests.
| HashFunction hf = Hashing.murmur3_128(); | ||
| Hasher topicHasher = hf.newHasher() | ||
| .putByte((byte) 0) // magic byte | ||
| .putLong(topicImage.id().hashCode()) // topic Id |
There was a problem hiding this comment.
Using hashCode() on a Uuid converts the 128-bit UUID into a 32-bit integer. This loses a significant amount of information and increases the likelihood of hash collisions. It's better to include the full UUID in the hash calculation by hashing its most and least significant bits.
| .putLong(topicImage.id().hashCode()) // topic Id | |
| .putLong(topicImage.id().getMostSignificantBits()) // topic Id | |
| .putLong(topicImage.id().getLeastSignificantBits()) |
| HashFunction hf = Hashing.murmur3_128(); | ||
| Hasher topicHasher = hf.newHasher() | ||
| .putByte((byte) 0) // magic byte | ||
| .putLong(FOO_TOPIC_ID.hashCode()) // topic Id |
There was a problem hiding this comment.
Using hashCode() on a Uuid is not ideal as it loses information from the 128-bit UUID, which increases the risk of collisions. It's better to hash the full 128 bits of the UUID. This change should be applied to all other places in this test file where hashCode() is used for a Uuid (lines 74, 91, 109).
| .putLong(FOO_TOPIC_ID.hashCode()) // topic Id | |
| .putLong(FOO_TOPIC_ID.getMostSignificantBits()) // topic Id | |
| .putLong(FOO_TOPIC_ID.getLeastSignificantBits()) |
| .filter(Optional::isPresent) | ||
| .map(Optional::get) |
There was a problem hiding this comment.
This stream operation can be simplified. Instead of filter(Optional::isPresent).map(Optional::get), you can use flatMap(Optional::stream) which is available since Java 9 and achieves the same result more concisely.
| .filter(Optional::isPresent) | |
| .map(Optional::get) | |
| .flatMap(Optional::stream) |
| HashFunction hf = Hashing.murmur3_128(); | ||
| Hasher topicHasher = hf.newHasher() | ||
| .putByte((byte) 0) // magic byte | ||
| .putLong(FOO_TOPIC_ID.hashCode()) // topic Id | ||
| .putString(FOO_TOPIC_NAME, StandardCharsets.UTF_8) // topic name | ||
| .putInt(FOO_NUM_PARTITIONS) // number of partitions | ||
| .putInt(0) // partition 0 | ||
| .putString("rack0;rack1", StandardCharsets.UTF_8) // rack of partition 0 | ||
| .putInt(1) // partition 1 | ||
| .putString("rack1;rack2", StandardCharsets.UTF_8); // rack of partition 1 |
There was a problem hiding this comment.
The logic for building the Hasher is duplicated across testComputeTopicHash, testComputeTopicHashWithDifferentMagicByte, testComputeTopicHashWithDifferentPartitionOrder, and testComputeTopicHashWithDifferentRackOrder. To improve maintainability, consider extracting this logic into a private helper method. This would reduce code duplication and make the tests easier to read and update.
Code Review: Hashing Logic and Performance👍 Well Done
📁 Selected files for review (5)
🎯 Custom Instructions
📝 Additional Comments
|
|
No review comments were generated during this review iteration of the PR. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (3)
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Group.java (2)
229-237: Enhance documentation for deterministic hash computation.The
computeGroupHashmethod lacks detailed documentation about:
- The purpose of this hash (e.g., for group assignment/routing)
- Why ordered hashing is important
- The stability guarantees across versions
Consider expanding the JavaDoc:
/** * Computes the hash of the topics in a group. * + * This method produces a deterministic hash by combining topic hashes in + * lexicographic order by topic name. The hash is stable as long as the + * topic set and individual topic hashes remain unchanged. + * * @param topicHashes The map of topic hashes. Key is topic name and value is the topic hash. * @return The hash of the group. */
256-264: Verify rack collection logic handles edge cases correctly.The rack collection logic filters nulls and empty optionals, but consider these edge cases:
- What if all brokers have null racks? Empty string is added to hash.
- What if a partition has no replicas? The inner stream would be empty.
- What if a replica ID doesn't exist in the cluster image? It's filtered by
filter(Objects::nonNull).The logic appears correct, but it would be valuable to have explicit test coverage for these edge cases.
Consider adding tests in GroupTest.java for:
- Topics with partitions having no racks
- Topics with partitions having empty replica sets
- Topics with partitions where some replicas don't exist in cluster image
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupTest.java (1)
163-188: Consider additional edge case tests for group hash.The current tests cover basic functionality and ordering, but consider adding tests for:
- Empty topic map (no topics)
- Single topic in the map
- Multiple topics with identical hash values (hash collision scenario)
- Large number of topics (performance/scalability check)
Example additional test:
@Test void testComputeGroupHashWithEmptyMap() { long result = Group.computeGroupHash(Map.of()); // Verify behavior with empty map - should this be allowed? } @Test void testComputeGroupHashWithSingleTopic() { long result = Group.computeGroupHash(Map.of(FOO_TOPIC_NAME, 456L)); // Verify single topic hashing works correctly }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (5)
build.gradle(1 hunks)checkstyle/import-control-group-coordinator.xml(1 hunks)gradle/dependencies.gradle(2 hunks)group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Group.java(2 hunks)group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupTest.java(1 hunks)
🧰 Additional context used
🧬 Code graph analysis (2)
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupTest.java (2)
clients/src/main/java/org/apache/kafka/common/Uuid.java (1)
Uuid(32-196)metadata/src/main/java/org/apache/kafka/image/MetadataImage.java (1)
MetadataImage(33-200)
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Group.java (3)
metadata/src/main/java/org/apache/kafka/image/ClusterImage.java (1)
ClusterImage(36-108)metadata/src/main/java/org/apache/kafka/image/TopicImage.java (1)
TopicImage(37-92)metadata/src/main/java/org/apache/kafka/metadata/BrokerRegistration.java (1)
BrokerRegistration(44-398)
🔇 Additional comments (4)
build.gradle (1)
1423-1423: LGTM - Dependency properly added.The Guava dependency is correctly added as an implementation dependency for the group-coordinator module.
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupTest.java (2)
51-117: Excellent test coverage for hash determinism.The tests comprehensively verify that:
- Magic byte changes affect the hash
- Partition order is significant
- Rack order within partitions is significant
- Different topic fields produce different hashes
This gives good confidence in the hash implementation's correctness.
45-48: MetadataImageBuilder is present in the codebase.The test utility class exists at
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/MetadataImageBuilder.java. No action needed.Likely an incorrect or invalid review comment.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Group.java (1)
247-247: Murmur3 is appropriate for metadata change detection.The
computeTopicHash()method hashes topic metadata (id, name, partition configuration, and rack assignments) to detect changes, not for partition assignment or group coordinator selection. For this metadata versioning use case, Murmur3 is appropriate—it's deterministic and sensitive to any input variation, making it effective for detecting whether topics have changed. Kafka correctly usesUtils.murmur2separately for partition key assignment and group coordinator selection, so there is no inconsistency with Kafka's standard hashing approach.
| classgraph: "4.8.173", | ||
| gradle: "8.10.2", | ||
| grgit: "4.1.1", | ||
| guava: "33.4.0-jre", |
There was a problem hiding this comment.
🧩 Analysis chain
Verify Guava version and consider lightweight alternatives.
Guava 33.4.0-jre is being added as a dependency. While Guava is a well-maintained library, it's quite large (~2.5 MB) and this PR only uses its hashing utilities. Consider:
- Verifying this is an appropriate version
- Evaluating whether a lighter-weight hashing library might suffice
Run the following script to check the Guava version and security status:
Additionally, consider searching for lightweight alternatives:
Also applies to: 151-151
🏁 Script executed:
#!/bin/bash
# Check latest Guava version
echo "=== Checking latest Guava version ==="
curl -s https://repo1.maven.org/maven2/com/google/guava/guava/maven-metadata.xml | grep -o '<version>[^<]*</version>' | tail -5
# Check for known vulnerabilities
echo -e "\n=== Checking for security advisories ==="
gh api graphql -f query='
{
securityVulnerabilities(first: 10, ecosystem: MAVEN, package: "com.google.guava:guava") {
nodes {
advisory {
summary
severity
publishedAt
}
vulnerableVersionRange
firstPatchedVersion {
identifier
}
}
}
}'Length of output: 972
🌐 Web query:
What are lightweight alternatives to Guava for Murmur3 hashing in Java?
💡 Result:
Here are small/lightweight Java alternatives to Guava’s Murmur3, with short notes and Maven/artifact references:
-
Apache Commons Codec — org.apache.commons:commons-codec — includes MurmurHash3 (32/128-bit) (small, well-maintained). (commons.apache.org) [1]
-
io7m jmurmur — com.io7m.jmurmur:jmurmur — pure-Java Murmur3 (32/64), minimal dependency footprint, on Maven Central. (io7m.com) [2]
-
Apache DataSketches (memory module) — org.apache.datasketches:datasketches-memory — MurmurHash3 implementation used by DataSketches (no Guava). Good if you already use DataSketches. (apache.github.io) [3]
-
eprst/murmur3 (github / Maven) — com.github.eprst:murmur3 — small, fast Murmur3 port (Unlicense). (mvnrepository.com) [4]
-
kotlinx-murmurhash / com.goncalossilva (Kotlin Multiplatform) — usable from JVM for MurmurHash3; lightweight if you accept a small KMP lib. (libraries.io) [5]
If you want, I can give example Maven coordinates + tiny usage snippet for any of these.
Citations:
- 1: https://commons.apache.org/proper/commons-codec/apidocs/org/apache/commons/codec/digest/MurmurHash3.html?utm_source=openai
- 2: https://www.io7m.com/software/jmurmur/?utm_source=openai
- 3: https://apache.github.io/datasketches-memory/4.0.0/org/apache/datasketches/memory/MurmurHash3.html?utm_source=openai
- 4: https://mvnrepository.com/artifact/com.github.eprst/murmur3/0.4.3?utm_source=openai
- 5: https://libraries.io/maven/com.goncalossilva%3Amurmurhash-js?utm_source=openai
🏁 Script executed:
#!/bin/bash
# Check if Guava is already in dependencies or newly added
echo "=== Searching for existing Guava usage in codebase ==="
rg -i "guava" --type gradle -C2 | head -20
echo -e "\n=== Checking for Murmur3 or hashing imports ==="
rg "Murmur3|guava.*hash|import.*guava" --type java -n | head -15
echo -e "\n=== Check gradle dependencies file structure ==="
head -80 gradle/dependencies.gradle | tail -30Length of output: 5248
🏁 Script executed:
#!/bin/bash
# Check JAR size of Guava 33.4.0-jre vs alternatives
echo "=== Guava dependency size comparison ==="
cat <<'EOF' | python3
import json
import urllib.request
libs = {
"Guava 33.4.0": "com.google.guava:guava:33.4.0-jre",
"Commons Codec 1.17": "commons-codec:commons-codec:1.17",
"io7m jmurmur": "com.io7m.jmurmur:jmurmur:7.0.0"
}
for name, coord in libs.items():
try:
# Query Maven central for artifact size
parts = coord.split(":")
group = parts[0].replace(".", "/")
artifact = parts[1]
version = parts[2]
url = f"https://repo1.maven.org/maven2/{group}/{artifact}/{version}/{artifact}-{version}.jar"
req = urllib.request.Request(url, method='HEAD')
with urllib.request.urlopen(req) as resp:
size_mb = int(resp.headers.get('Content-Length', 0)) / (1024 * 1024)
print(f"{name:25} {size_mb:6.2f} MB")
except Exception as e:
print(f"{name:25} [unavailable]")
EOFLength of output: 160
🏁 Script executed:
#!/bin/bash
# Find actual Guava imports (com.google.common.*)
echo "=== Actual Guava imports in Java source ==="
rg "import com\.google\.common" --type java -n | head -20
echo -e "\n=== Guava class usage (Hash, Hashing, etc.) ==="
rg "com\.google\.common\.(hash|base|collect)" --type java -C1 -n | head -25
echo -e "\n=== Check if Guava is new or existing by searching all gradle files ==="
rg "guava" gradle/ -nLength of output: 2611
Update Guava to the latest version (currently 33.5.0-jre).
Guava 33.4.0-jre is outdated. The codebase uses Guava's hashing API (com.google.common.hash.* for HashFunction, Hasher, HashCode, Hashing) in group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Group.java. While there are no active security vulnerabilities in version 33.4.0-jre, newer versions are available (33.4.8-jre as a patch, or 33.5.0-jre as the current release).
Consider upgrading to at least 33.4.8-jre or preferably 33.5.0-jre for bug fixes and improvements.
Regarding lightweight alternatives: while options like Apache Commons Codec exist for specific hash functions, they would not be direct drop-in replacements for Guava's full hashing API without refactoring the Group.java implementation.
| 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(";")); | ||
| topicHasher.putString(racks, StandardCharsets.UTF_8); // sorted racks with separator ";" | ||
| }); | ||
| return topicHasher.hash().asLong(); | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
Document hash versioning strategy and format stability.
The method uses a hardcoded magic byte (0) but lacks documentation about:
- How format changes will be handled if the hashing algorithm needs to evolve
- Whether this hash must remain stable across Kafka versions
- What guarantees are provided about hash stability
This is critical for understanding the long-term implications of this implementation.
Add clear documentation:
/**
* Computes the hash of the topic id, name, number of partitions, and partition racks by Murmur3.
*
+ * The hash format is versioned via a magic byte (currently 0). The hash includes:
+ * - Magic byte (0)
+ * - Topic ID hashCode
+ * - Topic name (UTF-8)
+ * - Partition count
+ * - For each partition (sorted): partition ID and semicolon-delimited sorted racks
+ *
+ * This hash is deterministic and stable for the same topic metadata. Changes to the
+ * hashing algorithm MUST use a different magic byte to maintain compatibility.
+ *
* @param topicImage The topic image.
* @param clusterImage The cluster image.
* @return The hash of the topic.
*/📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| 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(";")); | |
| topicHasher.putString(racks, StandardCharsets.UTF_8); // sorted racks with separator ";" | |
| }); | |
| return topicHasher.hash().asLong(); | |
| } | |
| /** | |
| * Computes the hash of the topic id, name, number of partitions, and partition racks by Murmur3. | |
| * | |
| * The hash format is versioned via a magic byte (currently 0). The hash includes: | |
| * - Magic byte (0) | |
| * - Topic ID hashCode | |
| * - Topic name (UTF-8) | |
| * - Partition count | |
| * - For each partition (sorted): partition ID and semicolon-delimited sorted racks | |
| * | |
| * This hash is deterministic and stable for the same topic metadata. Changes to the | |
| * hashing algorithm MUST use a different magic byte to maintain compatibility. | |
| * | |
| * @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(";")); | |
| topicHasher.putString(racks, StandardCharsets.UTF_8); // sorted racks with separator ";" | |
| }); | |
| return topicHasher.hash().asLong(); | |
| } |
🤖 Prompt for AI Agents
In group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Group.java
around lines 246 to 267, the computeTopicHash method uses a hardcoded magic byte
and lacks documentation about versioning, stability guarantees, and how to
evolve the hash format; add Javadoc above the method that documents the magic
byte meaning, the intended versioning strategy (e.g., bump magic byte when
changing field encoding or algorithm), whether the hash must be stable across
Kafka releases, and migration/compatibility expectations, and include a note
about charset, separator semantics, and that callers must not rely on numeric
stability unless explicitly guaranteed.
| 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(";")); | ||
| topicHasher.putString(racks, StandardCharsets.UTF_8); // sorted racks with separator ";" |
There was a problem hiding this comment.
🧩 Analysis chain
Potential issue with semicolon delimiter in rack names.
The rack collection logic joins rack names with semicolons (line 263). If a rack name contains a semicolon, this could cause ambiguity or unexpected hash values.
Consider either:
- Validating that rack names don't contain semicolons elsewhere in the codebase
- Using a different delimiter that's guaranteed not to appear in rack names
- Escaping semicolons in rack names before joining
Verify whether rack names can contain semicolons:
🏁 Script executed:
#!/bin/bash
# Search for rack name validation or constraints
rg -n "rack.*validation|rack.*allowed|rack.*pattern" --type java
# Search for rack name usage patterns
rg -n "rack\(\)" --type java -A 2 -B 2Length of output: 28037
🏁 Script executed:
#!/bin/bash
# Search for BrokerRegistrationRequest to see how rack is defined/validated
rg -n "class BrokerRegistrationRequest" --type java -A 20
# Search for where rack is validated or constrained
rg -n "rack.*setRack|setRack.*rack" --type java -B 3 -A 3
# Look for broker registration validation
rg -n "validateRack|validateBroker.*Rack|rack.*validate" --type javaLength of output: 80265
🏁 Script executed:
#!/bin/bash
# Search for rack in request/response data structures
fd -e java -x grep -l "rack" {} \; | grep -i "request\|response\|data" | head -20 | xargs rg -n "rack" -B 2 -A 2
# Search for any documentation or constraints on rack
rg -n "rack.*must|rack.*should|rack.*cannot|rack.*format" --type java -iLength of output: 63341
🏁 Script executed:
#!/bin/bash
# Search for BrokerRegistrationRequest schema or API definition
# This will help understand if rack has any character constraints at the protocol level
curl -s "https://raw.githubusercontent.com/apache/kafka/trunk/clients/src/main/java/org/apache/kafka/common/message/BrokerRegistrationRequest.json" 2>/dev/null | head -100Length of output: 14
🌐 Web query:
Kafka BrokerRegistration rack field constraints allowed characters
💡 Result:
Short answer: Kafka imposes no strict character whitelist for the broker "rack" value — it is treated as an opaque string (stored in broker registration/metadata and carried in protocol messages). Avoid using control characters and characters that would break any external systems you integrate with (e.g. JSON/URLs/filepaths, Kubernetes label values) but Kafka itself does not validate/limit the characters. [1][2][3]
Sources:
- Kafka protocol / metadata docs (rack is a string). [1]
- Broker registration schema (rack: "string", optional). [2]
- Community notes / discussion confirming rack is an opaque string (no enforced charset). [3]
Potential issue with semicolon delimiter in rack names.
The rack collection logic joins rack names with semicolons (line 263). If a rack name contains a semicolon, this could cause hash collisions or unexpected hash values (e.g., racks "a;b" and "a" + "b" would produce identical output).
While Kafka does not enforce character constraints on rack names at the protocol level, this delimiter choice creates a subtle risk if operators configure rack names containing semicolons.
Consider either:
- Validating that rack names don't contain semicolons at broker startup
- Using a different delimiter guaranteed not to appear in rack names (e.g., null byte or escaped format)
- Escaping semicolons in rack names before joining
🤖 Prompt for AI Agents
In group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Group.java
around lines 256 to 264 the code joins rack names with a semicolon which can
cause collisions or wrong hashes if a rack name contains ";" — fix by making the
serialization unambiguous: either validate/reject rack names containing ";" at
broker startup, or change the join to an escaped or length-prefixed form (e.g.,
escape ";" characters in rack names before joining or serialize each rack as
length + value, or join with a delimiter guaranteed not to appear such as the
null character), and update the topicHasher input accordingly so hashes are
stable and collision-free.
User description
Delete this text and replace it with a detailed description of your change. The
PR title and body will become the squashed commit message.
If you would like to tag individuals, add some commentary, upload images, or
include other supplemental information that should not be part of the eventual
commit message, please use a separate comment.
If applicable, please include a summary of the testing strategy (including
rationale) for the proposed change. Unit and/or integration tests are expected
for any behavior change and system tests should be considered for larger
changes.
CodeAnt-AI Description
Add deterministic topic and group hashes for metadata comparisons
What Changed
Impact
✅ Detects topic metadata mismatches regardless of ordering✅ Consistent group hashes when topic iteration order changes✅ Guarded hashing behavior with targeted unit tests💡 Usage Guide
Checking Your Pull Request
Every time you make a pull request, our system automatically looks through it. We check for security issues, mistakes in how you're setting up your infrastructure, and common code problems. We do this to make sure your changes are solid and won't cause any trouble later.
Talking to CodeAnt AI
Got a question or need a hand with something in your pull request? You can easily get in touch with CodeAnt AI right here. Just type the following in a comment on your pull request, and replace "Your question here" with whatever you want to ask:
This lets you have a chat with CodeAnt AI about your pull request, making it easier to understand and improve your code.
Example
Preserve Org Learnings with CodeAnt
You can record team preferences so CodeAnt AI applies them in future reviews. Reply directly to the specific CodeAnt AI suggestion (in the same thread) and replace "Your feedback here" with your input:
This helps CodeAnt AI learn and adapt to your team's coding style and standards.
Example
Retrigger review
Ask CodeAnt AI to review the PR again, by typing:
Check Your Repository Health
To analyze the health of your code repository, visit our dashboard at https://app.codeant.ai. This tool helps you identify potential issues and areas for improvement in your codebase, ensuring your repository maintains high standards of code health.
Summary by CodeRabbit
New Features
Tests
Dependencies
✏️ Tip: You can customize this high-level summary in your review settings.