在无索引的Kafka中搜索：为Kafdrop构建消息搜索的经验

消息在下午 2:14 离开第一个服务，却从未到达最后一个服务。在这两个事实之间：四个流水线阶段、六个主题、一个负责搬运每条记录的发布/订阅 API。值班工程师开着 Kafdrop，偏移量看起来正常，却没人能说清是哪个阶段吞掉了这条记录。这个缺口正是我们在黑客松上为 Kafdrop 构建消息搜索的原因。这个场景在我的日常工作中很熟悉：一个通过发布/订阅 API 在流水线阶段之间搬运数据的平台，一旦出问题，却没办法追问消息去了哪里。Kafdrop 能显示分区和偏移量，但无法搜索消息内部。Kafka 没有索引。这一句话就说清了全部问题。一个主题是按分区拆分的仅追加日志，消费者 API 按偏移量读取：给它 41,207，它就返回第 41,207 条记录。不存在“查找包含 order-88412 的记录”这种操作。要搜索，你只能选一个起点向前读，直到找到想要的内容或停下来。

这正是我们的功能所做的事，同时用护栏防止搜索伤害集群。有意思的部分是上限、消费者设置以及 SRE 约定。

为什么全量扫描不可行

最直接的做法是读取整个主题再 grep。主题里存着好几天数据，整体放不进内存。每次搜索都读取每条记录会冲击 broker，而在事故期间增加负载是最糟糕的时机。大多数搜索最终只有几条匹配，所以几乎所有这些工作都白费。

另一种方法是将每个主题镜像到搜索引擎中。这可行，很多团队也这么做。但这也意味着要再运行一套分布式系统，并且每次事故时都要解释索引延迟。对于一个黑客松调试工具来说，这套机器太重了。

设计：定位、扫描、停止

这个功能接收四个输入：主题、查询字符串、开始时间和两个上限。开始时间限定搜索回溯的范围。Kafka 的 offsetsForTimes 会按分区把时间戳映射到该时间点或之后的第一条偏移量；如果分区里没有这么新的数据，则返回 null。用户可以选择“最近一小时”或“自部署以来”。

结果上限约束内存：返回前 N 条匹配就停止。调试通常只需要一条记录，而不是一万条。

扫描上限约束 broker 负载：即使没有匹配，读取 M 条记录后也停止。无论用户输入什么，每次搜索都变成固定量的工作。一个拼写错误最多耗费 M 次读取。扫描量永远不会超过这个上限。轮询使用有界批次，因此工具一次最多只持有一页记录。任何操作都不会加载整个主题。

安全性：用 assign，不要用 subscribe

消费者设置正是搜索工具可能悄悄造成破坏的地方。有三个决定没有商量余地。

第一，搜索消费者使用 assign，而不是 subscribe。subscribe 会加入消费者组，带来协调和再均衡。调试工具不该掺和这些。assign 把固定的分区列表交给消费者，完全避开消费者组协调。

第二，消费者从不提交偏移量。自动提交关闭，代码也从不调用 commit。默认配置会自动提交，所以这一条最容易出问题：一个会推进生产消费者组偏移量的搜索工具就是恐怖故事，而修复只需一行配置。

第三，扫描上限同时充当安全控制：每次搜索最多读取 M 条记录，这是 SRE 在批准前能读懂的最坏情况。

搜索循环

以下是 Java 核心代码。三个常量构成了整个设计：要返回的匹配数、最坏情况下要读取的记录数、每次轮询的记录数。这些值只是示例；请根据你的主题进行调整。

$ java

private static final int RESULT_CAP = 25;
private static final int SCAN_CAP = 50_000;
private static final int POLL_BATCH = 500;

public List<FoundMessage> search(String topic, String query, long startTimeMs) {
    Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
    props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, String.valueOf(POLL_BATCH));

    List<FoundMessage> matches = new ArrayList<>();
    try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
        List<TopicPartition> partitions = new ArrayList<>();
        for (PartitionInfo p : consumer.partitionsFor(topic)) {
            partitions.add(new TopicPartition(topic, p.partition()));
        }
        consumer.assign(partitions);

        Map<TopicPartition, Long> times = new HashMap<>();
        for (TopicPartition tp : partitions) times.put(tp, startTimeMs);

        for (Map.Entry<TopicPartition, OffsetAndTimestamp> e : consumer.offsetsForTimes(times).entrySet()) {
            if (e.getValue() != null) consumer.seek(e.getKey(), e.getValue().offset());
            else consumer.seekToEnd(Collections.singletonList(e.getKey()));
        }

        long scanned = 0;
        while (scanned < SCAN_CAP && matches.size() < RESULT_CAP) {
            ConsumerRecords<String, String> batch = consumer.poll(Duration.ofMillis(500));
            if (batch.isEmpty()) break;
            for (ConsumerRecord<String, String> r : batch) {
                if (++scanned > SCAN_CAP) break;
                if (r.value() != null && r.value().contains(query)) {
                    matches.add(new FoundMessage(r.topic(), r.partition(), r.offset(), r.timestamp(), r.value()));
                    if (matches.size() >= RESULT_CAP) break;
                }
            }
        }
    }
    return matches;
}

从头看：配置关闭了自动提交，并固定了轮询批次大小。assign 接收一个固定的分区列表。offsetsForTimes 把开始时间转换为每个分区的起始偏移量；对于没有这么新数据的分区，null 分支会 seek 到末尾。循环向前读取，并在任一上限先触发时停止。没有任何提交，try-with-resources 会关闭消费者。匹配是对记录值做子串检查。墓碑消息携带 null，因此这个 null 检查至关重要。

交付：容器、Kubernetes 与 SRE 约定

应用以容器镜像形式交付：搜索服务及其配置，别无他物。它在 Kubernetes 上作为一个资源限制很紧的小型 Deployment 运行，因为结果上限已经约束了内存。一个副本就足够了。

$ config

apiVersion: apps/v1
kind: Deployment
metadata:
  name: kafdrop-search
spec:
  replicas: 1
  selector:
    matchLabels:
      app: kafdrop-search
  template:
    metadata:
      labels:
        app: kafdrop-search
    spec:
      containers:
        - name: search
          image: registry.example.com/kafdrop-search:1.0.0
          resources:
            requests:
              memory: "256Mi"
              cpu: "250m"
            limits:
              memory: "512Mi"
              cpu: "500m"
          env:
            - name: KAFKA_BROKERS
              value: "kafka:9092"

这份清单故意写得很无聊：一个副本、紧张的内存限制、broker 通过配置传入。和 SRE 的沟通很短，因为设计本身先回答了他们的疑问。这个工具只读：用 assign 而不是 subscribe，不提交偏移量，使用只读 Kafka 凭据。每次搜索都受扫描上限约束，所以最坏情况是一个所有人都能看懂的数字。这个约定比代码更重要。上限就是信任。

一套有效的调试流程

首先，用唯一标识在主题中搜索这条消息：订单 ID、请求 ID。找到了吗？记下偏移量和时间戳。

其次，检查下游阶段的消费者组延迟。如果消费者组已经消费到你这条消息的偏移量之后，说明该阶段读到了它，问题出在消费者的逻辑里。如果延迟卡住或持续增长，说明消费者落后、崩溃了，或者被一条毒消息卡住了。

第三，查看消费者：消息时间戳前后的日志、错误处理、死信路径。顺序很重要。它能把“消息从未到达”和“消息到达了但某处出了问题”区分开。以前，第一步靠猜；现在几分钟就能完成。

我们保留了什么

Kafka 提供的是顺序性和持久性，而不是搜索。要把搜索硬接到仅追加日志上，就意味着接受扫描。让它有界且无聊：定位到一个时间戳，以小批次向前读取，在第一个上限处停止。保留那三个让它安全的决定。用 assign 而不是 subscribe，这样工具永远不会加入消费者组。从不提交偏移量，这样读取不会改变任何东西。扫描上限把每次搜索变成已知量的 broker 工作。把这些做对，SRE 就会让你对生产环境运行它。做错了，你就造出了一种非常高效的、制造下一次事故的方法。

黑客松之后，团队保留了这个工具来做一件事：找出消息停在了哪里。一次搜索、一次延迟检查，再看一眼消费者，几分钟内完成。事故期间的诊断时间缩短了。

DZone 贡献者表达的观点仅代表其本人。

分类导航