Repository: apache/kafka
File: group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Utils.java (Java)

Code before:
     *
     * @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.
     * <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 UTF-8 encoded 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.

Review comment (suggestion):
I think we're adding a lot of unnecessary overhead for the hash computation (multiple map calls, etc.). We should probably just use an old school loop.

Code after:

        // 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 streaming XXH3.
     * <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>
     * For non-existent topics, the hash value is set to 0.
     * For existent topics, 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 with mostSignificantBits and leastSignificantBits.
     * 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 topicName     The topic image.
     * @param metadataImage The cluster image.
     * @return The hash of the topic.
     */
    static long computeTopicHash(String topicName, MetadataImage metadataImage) {
        TopicImage topicImage = metadataImage.topics().getTopic(topicName);
        if (topicImage == null) {
            return 0;
        }

        HashStream64 hasher = Hashing.xxh3_64().hashStream();
        hasher = hasher
            .putByte(TOPIC_HASH_MAGIC_BYTE)
            .putLong(topicImage.id().getMostSignificantBits())
            .putLong(topicImage.id().getLeastSignificantBits())
            .putString(topicImage.name())
            .putInt(topicImage.partitions().size());

        ClusterImage clusterImage = metadataImage.cluster();
        List<String> racks = new ArrayList<>();
        for (int i = 0; i < topicImage.partitions().size(); i++) {