如何查询Kafka指定Topic对应Key最新消息的offset/partition/timestamp

Kafka本身没有提供按消息Key直接点查的全局索引，不存在O(1)复杂度的查询方式，以下是生产验证可行的实现逻辑，覆盖临时排查、Java、Golang三类场景，最终可以拿到目标Key对应最新消息的partition、offset、timestamp信息。

核心实现逻辑 #

• 第一步：拉取目标Topic的全量分区列表，查询每个分区当前的最新写入位置（高水位值）
• 第二步：针对每个分区，从分区末尾反向批量遍历消息，匹配目标Key，找到该分区内对应Key的最新一条记录即可停止当前分区遍历，不需要扫完分区全量数据
• 第三步：汇总所有分区匹配到的记录，取时间戳最大的一条，就是该Key在全局维度的最新消息

注意：禁止从分区开头正向遍历，数据量大的Topic会产生极高的IO开销，查询延迟会达到分钟级甚至小时级。反向遍历的性能和Key的写入频率正相关，写入越频繁的Key查询越快。

方案1：命令行临时排查 #

适合小数据量Topic临时验证使用，依赖Kafka自带的控制台消费脚本：

# 替换变量为实际值：${BOOTSTRAP_SERVER}为Kafka连接地址，${TOPIC_NAME}为Topic名，${TARGET_KEY}为要查询的消息Key
kafka-console-consumer.sh \
  --bootstrap-server ${BOOTSTRAP_SERVER} \
  --topic ${TOPIC_NAME} \
  --from-beginning \
  --property print.key=true \
  --property print.value=false \
  --property print.timestamp=true \
  --property print.offset=true \
  --property print.partition=true \
  --key-deserializer org.apache.kafka.common.serialization.StringDeserializer \
  | grep "${TARGET_KEY}" | tail -n 1

输出结果从左到右依次为消息创建时间、分区、偏移量、消息Key，数据量超过10万条的Topic不建议用这个方案，优先用下面的代码实现反向遍历。

方案2：Java代码实现 #

使用官方kafka-clients客户端实现，不需要引入第三方依赖，逻辑稳定可靠。
首先引入Maven依赖：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.6.1</version>
</dependency>

核心实现代码：

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import java.time.Duration;
import java.util.*;

public class KafkaKeyQuery {
    /**
     * 查询指定Key对应的最新消息
     * @return 匹配到的最新消息，无匹配结果返回null
     */
    public static ConsumerRecord<byte[], byte[]> getLatestRecordByKey(String bootstrapServers, String topic, byte[] targetKey) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "500"); // 单批拉取数量，可根据单条消息大小调整

        try (KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(props)) {
            // 获取Topic全部分区
            List<PartitionInfo> partitionMeta = consumer.partitionsFor(topic);
            List<TopicPartition> tps = partitionMeta.stream()
                    .map(p -> new TopicPartition(topic, p.partition()))
                    .toList();
            consumer.assign(tps);

            // 查询每个分区的最新写入偏移量
            Map<TopicPartition, Long> endOffsetMap = consumer.endOffsets(tps);
            ConsumerRecord<byte[], byte[]> latestRecord = null;

            for (TopicPartition tp : tps) {
                long currentSeekPos = endOffsetMap.get(tp) - 1;
                boolean foundInPartition = false;
                // 从分区末尾向前逐批扫描
                while (currentSeekPos >= 0 && !foundInPartition) {
                    long batchStart = Math.max(0, currentSeekPos - 500 + 1);
                    consumer.seek(tp, batchStart);
                    ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofSeconds(3));
                    List<ConsumerRecord<byte[], byte[]>> tpRecords = records.records(tp);
                    // 倒序遍历当前批次，第一个匹配Key的就是该分区内的最新对应消息
                    for (int i = tpRecords.size() - 1; i >= 0; i--) {
                        ConsumerRecord<byte[], byte[]> record = tpRecords.get(i);
                        if (Arrays.equals(record.key(), targetKey)) {
                            if (latestRecord == null || record.timestamp() > latestRecord.timestamp()) {
                                latestRecord = record;
                            }
                            foundInPartition = true;
                            break;
                        }
                    }
                    currentSeekPos = batchStart - 1;
                }
            }
            return latestRecord;
        }
    }

    public static void main(String[] args) {
        ConsumerRecord<byte[], byte[]> result = getLatestRecordByKey(
                "localhost:9092",
                "order_topic",
                "order_10086".getBytes()
        );
        if (result != null) {
            System.out.printf("查询结果：partition=%d, offset=%d, timestamp=%d%n",
                    result.partition(), result.offset(), result.timestamp());
        } else {
            System.out.println("未找到对应Key的消息");
        }
    }
}

方案3：Golang代码实现 #

使用生产环境广泛采用的confluent-kafka-go客户端（基于librdkafka，性能优异），逻辑和Java版本保持一致。
首先安装依赖：

go get github.com/confluentinc/confluent-kafka-go/kafka

核心实现代码：

package main

import (
	"bytes"
	"fmt"
	"github.com/confluentinc/confluent-kafka-go/kafka"
	"time"
)

// QueryLatestByKey 查询指定Key对应的最新消息，无匹配返回nil
func QueryLatestByKey(bootstrapServers, topic string, targetKey []byte) (*kafka.Message, error) {
	consumer, err := kafka.NewConsumer(&kafka.ConfigMap{
		"bootstrap.servers": bootstrapServers,
		"group.id":          fmt.Sprintf("tmp-query-%d", time.Now().UnixMilli()), // 用临时消费组，避免影响业务
		"auto.offset.reset": "earliest",
		"enable.auto.commit": false,
		"fetch.max.bytes":    1024 * 1024 * 2, // 单批拉取最大2M，可按需调整
	})
	if err != nil {
		return nil, err
	}
	defer consumer.Close()

	// 获取Topic全部分区元数据
	meta, err := consumer.GetMetadata(&topic, false, 5000)
	if err != nil {
		return nil, err
	}
	topicMeta := meta.Topics[topic]
	var tps []kafka.TopicPartition
	partitionEndOffset := make(map[int32]int64)
	for _, p := range topicMeta.Partitions {
		tp := kafka.TopicPartition{Topic: &topic, Partition: p.ID}
		tps = append(tps, tp)
		// 查询分区高水位
		high, _, err := consumer.QueryWatermarkOffsets(topic, p.ID, 5000)
		if err != nil {
			return nil, err
		}
		partitionEndOffset[p.ID] = high - 1 // 高水位是下一条消息的写入位置，最新消息偏移量为high-1
	}

	if err = consumer.Assign(tps); err != nil {
		return nil, err
	}

	var latestMsg *kafka.Message
	batchSize := int64(500) // 单批拉取条数，可根据单条消息大小调整

	for pId, endOffset := range partitionEndOffset {
		currentPos := endOffset
		found := false
		tp := kafka.TopicPartition{Topic: &topic, Partition: pId}
		for currentPos >= 0 && !found {
			batchStart := currentPos - batchSize + 1
			if batchStart < 0 {
				batchStart = 0
			}
			if err = consumer.Seek(tp, batchStart, 5000); err != nil {
				return nil, err
			}
			// 拉取当前批次消息
			var batch []*kafka.Message
			needReadCnt := int(currentPos - batchStart + 1)
			timeout := time.After(3 * time.Second)
			for len(batch) < needReadCnt {
				select {
				case ev := <-consumer.Events():
					switch e := ev.(type) {
					case kafka.Message:
						batch = append(batch, &e)
					case kafka.Error:
						return nil, e
					}
				case <-timeout:
					break
				}
			}
			// 倒序匹配Key
			for i := len(batch) - 1; i >= 0; i-- {
				msg := batch[i]
				if bytes.Equal(msg.Key, targetKey) {
					if latestMsg == nil || msg.Timestamp.UnixMilli() > latestMsg.Timestamp.UnixMilli() {
						latestMsg = msg
					}
					found = true
					break
				}
			}
			currentPos = batchStart - 1
		}
	}
	return latestMsg, nil
}

func main() {
	msg, err := QueryLatestByKey("localhost:9092", "order_topic", []byte("order_10086"))
	if err != nil {
		panic(err)
	}
	if msg != nil {
		fmt.Printf("查询结果：partition=%d, offset=%d, timestamp=%d\n",
			msg.TopicPartition.Partition, msg.TopicPartition.Offset, msg.Timestamp.UnixMilli())
	} else {
		fmt.Println("未找到对应Key的消息")
	}
}

优化注意事项 #

• 如果Topic开启了日志压缩（Log Compaction）策略，每个分区内同一个Key只会保留最新版本，扫描时只要匹配到Key就可以直接停止当前分区遍历，不需要再往前扫，查询性能会大幅提升
• 单批拉取的条数/大小可以根据单条消息平均大小调整，单条消息越大，批次值调得越小，避免单次拉取占用过多带宽
• 如果业务使用自定义时间戳而非Broker默认的写入时间，跨分区比对最新记录时，需要替换成业务自定义的新老判定逻辑，不同分区的offset值没有可比性，不能直接作为跨分区消息新老的判断依据
• 临时查询使用随机生成的消费组ID即可，不要复用业务消费组ID，避免影响业务消费进度

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