Kafka技术咨询：如何检查消息Key对应分区及验证分配规则

嘿，这个问题我之前在调试Kafka消息一致性的时候也折腾过，刚好把经验分享给你～

一、计算指定Key会被发送到哪个分区 #

Kafka默认使用DefaultPartitioner处理带Key的消息路由，核心逻辑很直接：

1. 对消息Key做Murmur2哈希运算（这是Kafka内置的低碰撞率哈希算法）
2. 把哈希值对Topic的总分区数取模，得到的结果就是目标分区编号（分区从0开始计数）

如果你想自己验证这个逻辑，可以用简单的代码实现，比如Java版本：

import org.apache.kafka.common.utils.Utils;

public class KeyPartitionCalculator {
    public static int calculatePartition(String key, int totalPartitions) {
        if (key == null) {
            return -1; // 无Key时走轮询，这里我们只关注有Key的场景
        }
        byte[] keyBytes = key.getBytes();
        int hash = Utils.murmur2(keyBytes);
        return Math.abs(hash) % totalPartitions;
    }
}

或者Python版本（模拟Kafka原生Murmur2实现）：

def murmur2(key):
    m = 0x5bd1e995
    r = 24
    seed = 0x9747b28c
    length = len(key)
    h = seed ^ length
    data = bytearray(key.encode('utf-8'))
    
    while len(data) >= 4:
        k = data[0] | (data[1] << 8) | (data[2] << 16) | (data[3] << 24)
        k &= 0xffffffff
        k *= m
        k &= 0xffffffff
        k ^= k >> r
        k &= 0xffffffff
        k *= m
        k &= 0xffffffff
        h *= m
        h &= 0xffffffff
        h ^= k
        data = data[4:]
    
    if len(data) >= 1:
        h ^= data[-1] << (8 * (len(data)-1))
        h &= 0xffffffff
        h *= m
        h &= 0xffffffff
    
    h ^= h >> 13
    h &= 0xffffffff
    h *= m
    h &= 0xffffffff
    h ^= h >> 15
    h &= 0xffffffff
    return h

def calculate_partition(key, total_partitions):
    if not key:
        return -1
    hash_val = murmur2(key)
    return abs(hash_val) % total_partitions

注意：如果你的Topic用了自定义分区器，那就要按照自定义逻辑计算，上面的代码只适用于默认分区策略。

二、检查Key已分配到的分区 #

如果已经发送了消息，或者想验证实际路由结果，可以用以下几种方法：

1. 命令行工具快速验证 #

• 发送带Key的消息：用kafka-console-producer.sh指定Key格式发送

  kafka-console-producer.sh --bootstrap-server <kafka-broker>:9092 --topic <your-topic> --property parse.key=true --property key.separator=:
  

  输入test-key:test-value即可发送一条带Key的消息。

• 消费时打印分区信息：用kafka-console-consumer.sh消费并显示分区号

  kafka-console-consumer.sh --bootstrap-server <kafka-broker>:9092 --topic <your-topic> --from-beginning --property print.partition=true --property print.key=true
  

  输出会类似：[Partition 2] test-key: test-value，这里的Partition 2就是该Key对应的分区。

2. 消费端代码打印分区 #

在消费逻辑里直接输出消息的分区信息（以Java为例）：

consumer.subscribe(Collections.singletonList("your-topic"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("Key: %s, 所属分区: %d, Value: %s%n", 
            record.key(), record.partition(), record.value());
    }
}

3. 解析Kafka日志文件 #

如果消息已经持久化到Broker，可以用kafka-dump-log.sh解析对应分区的日志文件：

kafka-dump-log.sh --files /path/to/kafka/logs/your-topic-2/00000000000000000000.log --print-data-log

日志文件名里的your-topic-2中的2就是分区号，输出内容会包含每条消息的Key和分区归属。

内容的提问来源于stack exchange，提问作者Jal