Searching Kafka Without an Index: Lessons From Building Message Search for Kafdrop

The message left the first service at 2:14 PM and never reached the last one. Between those facts: four pipeline stages, six topics, a pub/sub API moving every record. The engineer on call had Kafdrop open, the offsets looked fine, and nobody could say which stage had eaten the record.

That gap is why we built message search for Kafdrop at a hackathon. The setup was familiar from my day job: a platform moving data between pipeline stages over a pub/sub API, with no way to ask where a message went when something breaks. Kafdrop showed partitions and offsets. It couldn't search inside the messages.

Kafka has no index. That is the whole problem in one sentence. A topic is an append-only log split across partitions, and the consumer API reads by offset: hand it 41,207, and it returns record 41,207. There is no "find the record containing order-88412." To search, you pick a starting point and read forward until you find what you want or stop.

So that is what our feature does, with guardrails that keep the search from hurting the cluster. The interesting part was the caps, the consumer setup, and the SRE agreement.

Why a Full Scan Is Not an Option

The obvious approach is to read the whole topic and grep. Topics hold days of data, so the whole thing doesn't fit in memory. Reading every record on every search hammers the brokers, and during an incident that is the worst time to add load. Most searches end with a few matches, so nearly all of that work is wasted.

The other approach is to mirror every topic into a search engine. That works, and plenty of teams do it. It also means running a second distributed system and explaining index lag during every incident. For a hackathon debugging tool, that was too much machinery.

The Design: Seek, Scan, and Stop

The feature takes four inputs: a topic, a query string, a start time, and two caps.

The start time bounds how far back the search goes. Kafka's offsetsForTimes maps a timestamp to the first offset at or after it per partition, or null when the partition holds nothing that new. The user picks "the last hour" or "since the deploy."

The result cap bounds memory: return the first N matches and stop. Debugging usually needs one record, not ten thousand.

The scan cap bounds broker load: stop after M records even with no match. Every search becomes a fixed amount of work, whatever the user types. A typo costs M reads. The scan can never grow beyond that.

Polls use bounded batches, so the tool never holds more than one page of records at a time. Nothing loads a whole topic.

Safety: Assign, Don't Subscribe

The consumer setup is where a search tool can quietly cause damage. Three decisions were non-negotiable.

First, the search consumer uses assign, not subscribe. Subscribe joins a consumer group, bringing coordination and rebalances. A debugging tool has no business in any of that. Assign hands the consumer a fixed partition list and stays out of group coordination entirely.

Second, the consumer never commits offsets. Auto-commit is off, and the code never calls commit. The default config commits automatically, so this is the one that bites: a search tool that advances a production group's offsets is a horror story, and the fix is one config line.

Third, the scan cap doubles as a safety control: every search reads at most M records, a worst case SRE can read before approving.

The Search Loop

Here is the core in Java. The three constants are the whole design: matches to return, records to read in the worst case, records per poll. The values are illustrative; tune them to your topics.

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;
}

From the top: the config disables auto-commit and fixes the poll batch size. Assign takes a fixed partition list. offsetsForTimes turns the start time into a starting offset per partition; the null branch seeks to the end for partitions with nothing that new. The loop reads forward and stops at whichever cap hits first. Nothing commits, and try-with-resources closes the consumer.

Matching is a substring check on the record value. Tombstones carry nulls, so the null check is load-bearing.

Shipping It: A Container, Kubernetes, and the SRE Agreement

The app ships as a container image: the search service and its config, nothing else. It runs on Kubernetes as a small deployment with tight resource limits, since the result cap already bounds memory. One replica is plenty.

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"

The manifest is deliberately boring: one replica, tight memory limits, brokers passed as config.

The SRE conversation was short, because the design answered their questions first. The tool only reads: assign instead of subscribe, no offset commits, read-only Kafka credentials. Every search is bounded by the scan cap, so the worst case was a number everyone could read.

That agreement mattered more than the code. The caps were the trust.

A Debugging Routine That Works

First, search the topic for the message using something unique: an order ID, a request ID. Found it? Note the offset and timestamp.

Second, check consumer group lag for the downstream stage. If the group consumed past your message's offset, the stage read it, and the problem is in the consumer's logic. If lag is stuck or growing, the consumer is behind, crashed, or wedged on a poison record.

Third, look at the consumer: logs around the message's timestamp, error handling, the dead-letter path.

The order matters. It separates "the message never arrived" from "the message arrived and something went wrong." Before, step one was guesswork. Now it takes minutes.

What We Kept

Kafka gives you ordering and durability, not search. Bolting search onto an append-only log means accepting the scan. Make it bounded and boring: seek to a timestamp, read forward in small batches, stop at the first cap.

Keep the three decisions that made it safe. Assign instead of subscribe, so the tool never joins a consumer group. Never commit offsets, so reading changes nothing. A scan cap that turns every search into a known quantity of broker work. Get those right, and SRE will let you run it against production. Get them wrong, and you have built a very efficient way to cause your next incident.

After the hackathon, the team kept the tool for one job: finding where a message stopped. A search, a lag check, and a look at the consumer, done in minutes. Diagnosis time during incidents fell.