1.8 million messages, 200 million rows: checking Kafka against ClickHouse

One message, a hundred rows

Our screens send proof-of-play in batches. Every upload from a player becomes one Kafka message with all the lines of that upload, separated by newlines. In a small sample I read on October 5, the median message of the main audit topic had 12.7 KB and 112 lines. That is why about 200 million rows a day travel in less than 2 million messages.

We designed this pipeline together in the team; I prototyped and tested the approach, and a teammate wrote and shipped the final consumer. The check in this note is mine. I wanted two answers that the dashboards didn't give. How many rows does a message really carry? And does what goes into Kafka come out in ClickHouse, row by row?

Counting messages without consuming

Offsets already have the answer. Ask for the offset of the first message after "24 hours ago", ask for the current end of each partition, and subtract. It reads metadata only, no message, no commit, so it doesn't disturb the real consumers:

import time
from confluent_kafka import Consumer, TopicPartition

c = Consumer({
    "bootstrap.servers": BROKERS,
    "group.id": "readonly-count-tmp",     # throwaway, never subscribes
    "enable.auto.commit": False,
})

def messages_last_24h(topic: str) -> int:
    parts = c.list_topics(topic, timeout=10).topics[topic].partitions
    since = int((time.time() - 86_400) * 1000)
    start = c.offsets_for_times([TopicPartition(topic, p, since) for p in parts], timeout=10)
    total = 0
    for tp in start:
        _low, high = c.get_watermark_offsets(TopicPartition(topic, tp.partition), timeout=10)
        total += high - (tp.offset if tp.offset >= 0 else high)   # -1 = nothing newer
    return total

To see the size of a message I read one message per partition, 19 in total, with assign() in a throwaway group and no commit, and I only measured bytes and counted newlines. Nothing of the content was printed or saved. It's a small sample, so the 112 is an estimate, and I say it like that.

Counting rows on the other side

On the ClickHouse side count() on the table is the wrong tool. Our audit tables are ReplacingMergeTree, so count() changes when parts merge, and on a table with a TTL old rows also leave. What I want is what was inserted, and system.part_log records each new part with its row count at the moment of the insert:

SELECT table, sum(rows) AS rows_inserted
FROM system.part_log
WHERE event_type = 'NewPart'               -- inserts only; merges are MergeParts
  AND database = 'audit'
  AND table LIKE 'play_log%'
  AND event_time >= now() - INTERVAL 24 HOUR
GROUP BY table;

The numbers

For telemetry, where one message is one row, the comparison is direct: 13,912,374 messages in the presence topics against 13,912,522 rows in the ClickHouse table, in the same 24 hours, from 13,835 screens. A difference of 148 rows, 0.001%. My reading is that the two windows can't start in the exact same millisecond (I think that's all it is), and I didn't chase it further.

Proof-of-play had 1,782,599 messages in the main topic in 24 hours, and part_log showed 193,924,874 rows inserted in the main play tables plus 5,380,579 in the smaller environment, about 199.3 million. Multiply 1.78 million messages by ~112 rows and you land close to that, which is as much as a sample of six messages can tell.

All the main topics together had about 22.5 million messages that day, and that one is a floor, since retention cuts part of the shortest topics. Retention is set per type of data: 3 hours for telemetry that only passes through, 3 days for proof-of-play. The 3 days matter. Once the importer stopped for about 39 hours and the messages just waited in the topic until it came back.

Why I check this way

A lag graph tells if consumers follow the offsets. Offsets against part_log tell if the data arrived, and in what quantity. I had used the same part_log idea in August to check a summary table against its source, and I keep going back to it because it is cheap, it reads metadata, and it gives a number that someone else can repeat.

Filed under the problem it solves

• kafka
• clickhouse
• part_log
• data validation
• observability
• batching