Kafka and ClickHouse

ClickHouse reads a Kafka topic through the Kafka table engine, which is a thin wrapper around a Kafka consumer. That is why almost every production surprise with it is a Kafka behaviour wearing a ClickHouse costume: lag that grows on one partition, a rebalance loop, a replay that duplicates rows, a topic that quietly deleted the data you were about to read. This guide covers the Kafka model you need to reason about those, builds the standard pipeline, puts numbers on consumer and partition sizing, and ends with the operations checklist.

Everything here that needs a broker could not be run on the machine used to write it. The ClickHouse DDL was created on ClickHouse 26.4 (the engine accepts the definitions without connecting) and re-checked with clickhouse format; no messages were consumed, so treat output descriptions as explanations, and test the pipeline against a real topic before relying on it.

Contents

• What the Kafka engine is, and is not
• Kafka concepts that matter to ClickHouse users
• The pipeline: Kafka table, materialized view, MergeTree
• Consumer and partition math
• Bad messages: skip, stream, dead letter
• Measuring lag
• Rebalances
• Retention, cleanup policy and the recovery window
• Pausing, changing and replaying
• Operations checklist
• Kafka CLI cheat sheet

What the Kafka engine is, and is not

A Kafka table does not store anything. It is a consumer you can SELECT from, and a SELECT consumes the messages it returns: run it by hand and those messages are gone from the group’s point of view. The standard way to get data in is therefore three objects working together:

1. a Kafka table that knows how to read the topic,
2. a MergeTree table that stores the rows,
3. a materialized view that continuously moves rows from the first to the second.

The view is the thing that runs the consumer. Everything the view guide says about incremental views applies unchanged: the view sees one block at a time, a failing view fails the block, and the target is written outside any transaction.

Kafka concepts that matter to ClickHouse users

Topics and partitions. A topic is a named append-only log, roughly the counterpart of a table. It is split into partitions, and six rules cover most of what matters: a topic has at least one partition, numbered from 0; every record belongs to exactly one partition; every record in a partition has an offset, a sequence number starting at 0; order is guaranteed only inside a partition, never across the whole topic; records are immutable and old data disappears by retention; and records with the same key land in the same partition as long as the partition count does not change.

The fourth rule is the one that bites ClickHouse users. If you need “the latest event per entity”, have the producer use the entity id as the key, so all of its events share a partition and keep their order. Across partitions, ClickHouse will see interleaved data in whatever order consumers happen to read it.

Brokers, replication and leaders. A broker is one Kafka server; connect to any one (the bootstrap broker) and you learn about the rest. Each partition is stored as replication.factor copies, one leader that serves reads and writes and followers that copy from it; the caught-up ones are the in-sync replicas. Current Kafka (4.x) runs on its own Raft-based controller and has no ZooKeeper, which is why client tools take --bootstrap-server and the old --zookeeper flag is gone.

Producers. A producer picks the partition by hashing the key (round-robin without one) and sends to that partition’s leader. acks=0 is fire and forget, acks=1 means the leader wrote it (lost if the leader dies before followers copy), acks=all means every in-sync replica has it and pairs with min.insync.replicas on the topic. Two producer settings explain most “why is my data late” questions: linger.ms, how long the producer waits to fill a batch (default 0), and batch.size. For retries without duplicates, enable.idempotence=true.

Consumers and groups. Consumers pull. Each belongs to a group with a group id, and the cluster stores one committed offset per group, topic and partition. Inside a group each partition is assigned to exactly one consumer; different groups are independent and each sees every message. That is how a second ClickHouse table can read the same topic without disturbing the first: give it another group name. The consequence for sizing is blunt: more consumers than partitions leaves the extras idle.

Delivery semantics. The moment the offset is committed decides the guarantee. Commit first and process after is at most once (a crash loses the batch); process first and commit after is at least once (a crash redelivers it); exactly once exists only inside Kafka, between Kafka topics. ClickHouse commits after the batch has been pushed to the dependent materialized views, which gives at-least-once: after a crash or a rebalance a block can be read again. Plan for duplicates.

Offsets expire. Committed offsets are kept for offsets.retention.minutes after a group goes idle (seven days by default in current Kafka). After that the group’s position is forgotten and the consumer falls back to auto.offset.reset: latest skips everything produced meanwhile, earliest re-reads the whole retained log. A table that was detached for a long time can therefore silently skip or replay data when it is attached again.

The pipeline: Kafka table, materialized view, MergeTree

Start with the table that holds the data. Besides the demo columns it keeps the message coordinates, a few bytes per row that buy traceability (which message did this row come from?) and a deduplication handle after a replay:

CREATE TABLE analytics.events
(
    event_time DateTime,
    user_id UInt64,
    event_type LowCardinality(String),
    url String,
    duration_ms UInt32,
    country LowCardinality(String),
    kafka_topic LowCardinality(String),
    kafka_partition UInt32,
    kafka_offset UInt64
)
ENGINE = MergeTree
ORDER BY (event_type, event_time);

The queue table declares the same payload columns. Four settings are required: brokers, topic, consumer group and message format.

CREATE TABLE analytics.events_queue
(
    event_time DateTime,
    user_id UInt64,
    event_type LowCardinality(String),
    url String,
    duration_ms UInt32,
    country LowCardinality(String)
)
ENGINE = Kafka
SETTINGS
    kafka_broker_list = 'kafka-1:9092,kafka-2:9092,kafka-3:9092',
    kafka_topic_list = 'events',
    kafka_group_name = 'clickhouse-events-v1',
    kafka_format = 'JSONEachRow',
    kafka_num_consumers = 3,
    kafka_thread_per_consumer = 1,
    kafka_handle_error_mode = 'stream';

kafka_group_name is the Kafka consumer group: two tables with the same name split the partitions, two with different names each get everything. Put a version in it (-v1) so that you can start a fresh consumer later without touching the old offsets. kafka_format is any ClickHouse input format; JSONEachRow, Avro, Protobuf and CSV are all used. The block-size settings (kafka_max_block_size, kafka_poll_max_batch_size, kafka_flush_interval_ms) decide how many messages go into one block, and a block becomes one inserted part, so they matter for too many parts; the next section puts numbers on them.

The view starts consumption when it is created. Use TO, so that the target can be altered, backed up and queried independently of the view:

CREATE MATERIALIZED VIEW analytics.events_mv TO analytics.events AS
SELECT event_time, user_id, event_type, url, duration_ms, country,
       _topic AS kafka_topic, _partition AS kafka_partition, _offset AS kafka_offset
FROM analytics.events_queue
WHERE length(_error) = 0;

The SELECT is where you transform: cast types, parse JSON strings, drop fields, derive columns. One queue table can feed several views and every view receives every block. The engine’s virtual columns are _topic, _key, _offset, _timestamp, _timestamp_ms, _partition and _headers.name / _headers.value; they are not in DESCRIBE but a view can select them. With kafka_handle_error_mode = 'stream' there are two more, _raw_message and _error, filled only when parsing failed. The WHERE length(_error) = 0 is not decoration: a message that failed to parse still arrives as a row, with its payload columns empty or defaulted, and without the filter it would be inserted into the target as a row of zeros.

(_topic, _partition, _offset) identifies a message uniquely, so it is the natural deduplication key. Two caveats about using it. A ReplacingMergeTree keyed on it would put offsets in the sorting key and lose the time-based primary index that queries want, so most pipelines keep the coordinates as plain columns for audit and deduplicate on a business id carried in the message (ORDER BY (event_type, event_time, event_id) on a ReplacingMergeTree) instead. And deduplication in a ReplacingMergeTree is eventual: until two rows meet in a merge, both exist, and you read with FINAL or argMax when exactness matters.

Consumer and partition math

Sizing has four numbers. P is the partition count of the topic, C is kafka_num_consumers on each table, R is the number of ClickHouse servers (replicas) reading the same group, and the total consumer count is R × C. Consumers beyond P get no partition and sit idle, and the Kafka engine documentation also says the total should not exceed the number of physical cores (there is a kafka_disable_num_consumers_limit setting that removes the core-based cap, which exists to be used rarely).

Topic partitions (P)	Servers (R)	Consumers per table (C)	Total	Result
12	1	12	12	one partition each; needs 12 cores
12	3	4	12	one partition each, spread over three servers
12	3	6	18	six consumers idle
12	4	4	16	four idle; adding the fourth server bought nothing
12	3	3	9	three consumers own two partitions; the others own one

The last row shows the part people miss: when P is not a multiple of the consumer count, some consumers carry more partitions than others, and the group is as slow as its busiest member. Pick P with the consumer count in mind, and leave headroom, because adding partitions later changes the key-to-partition mapping and breaks per-key ordering for data already in the topic. Partitions are not free either (open files, memory and leader elections on the brokers); a small multiple of the broker count is a common start, with a replication factor of 3 in production.

The second effect of C is on parts. Messages are grouped into blocks, and a block is flushed when it reaches kafka_max_block_size rows (default: the value of max_insert_block_size, 1,048,449 rows on 26.4) or when kafka_flush_interval_ms elapses (default: stream_flush_interval_ms, 7,500 ms), whichever comes first. With kafka_thread_per_consumer = 1 every consumer flushes independently; without it, the rows of several consumers are squashed into one block. Each flushed block is one insert into the target, so each flush creates at least one part (one per partition-key value the block touches).

Worked arithmetic, using the defaults: twelve independent consumers on a topic that is quiet enough for the time limit to win flush 12 × (60 ÷ 7.5) = 96 blocks a minute, 138,240 a day, every one a part until merged. The target’s limits are parts_to_delay_insert = 1,000 and parts_to_throw_insert = 3,000 active parts in one partition on 26.4 (read from system.merge_tree_settings), so the system stays healthy only because merges consume parts faster than the consumers create them. If you raise kafka_flush_interval_ms to 30,000 the same consumers make 24 parts a minute, at the cost of data that is up to 30 seconds older when it reaches the table. If a consumer is fast enough to fill the 1,048,449-row block before the timer, the block size rules instead: a consumer reading 50,000 rows/s needs about 21 s to fill one, so the timer wins and blocks hold about 375,000 rows. These are the numbers to plug your own rates into, and the reason kafka_thread_per_consumer = 1 is not free: it trades parts for parallel flushing.

So the first lever for a pipeline that makes too many parts is not fewer partitions in the topic but a longer flush interval, bigger blocks and, if the target’s partition key puts every consumer’s block into the same partition, fewer independent flushers.

Bad messages: skip, stream, dead letter

By default a message that fails to parse raises an exception, and the consumer retries the same block forever. One poison message stalls the whole pipeline, and the symptom is lag growing on exactly one partition. There are three levels of defence.

kafka_skip_broken_messages = N tolerates up to N unparsable messages per block and drops them silently. It keeps the pipeline moving and loses the evidence; use it only for data you can afford to lose.

kafka_handle_error_mode = 'stream' keeps the failed messages. The raw message and the parse error land in the virtual columns _raw_message and _error, and you route them with a second view into a quarantine table (the main view above already filters them out):

CREATE TABLE analytics.events_errors
(
    ts DateTime DEFAULT now(),
    kafka_topic LowCardinality(String),
    kafka_partition UInt32,
    kafka_offset UInt64,
    error String,
    raw String
)
ENGINE = MergeTree
ORDER BY (ts, kafka_topic, kafka_partition, kafka_offset)
TTL ts + INTERVAL 30 DAY;

CREATE MATERIALIZED VIEW analytics.events_errors_mv TO analytics.events_errors AS
SELECT _topic AS kafka_topic, _partition AS kafka_partition, _offset AS kafka_offset,
       _error AS error, _raw_message AS raw
FROM analytics.events_queue
WHERE length(_error) > 0;

Both views read the same blocks, so each message ends up in exactly one of the two tables, with its Kafka coordinates, and the 30-day TTL keeps the quarantine from growing for ever. The pipeline never stalls on a bad payload and you can replay the quarantined messages after fixing the producer or the schema.

kafka_handle_error_mode = 'dead_letter_queue' writes failed messages into system.dead_letter_queue instead, which saves the second table and view. That system table exists only when the server configuration contains a dead_letter_queue section (database, table, partition strategy and flush interval, in the same shape as the other system log tables; check your version’s configuration reference for the exact keys), and it is implemented for Kafka and RabbitMQ. Its columns include event_time, table, error, raw_message, kafka_topic_name, kafka_partition, kafka_offset and kafka_key:

SELECT event_time, `table`, kafka_topic_name, kafka_partition, kafka_offset, error, raw_message
FROM system.dead_letter_queue
WHERE `table` = 'events_queue'
ORDER BY event_time DESC
LIMIT 20;

For anything you intend to reprocess, the explicit quarantine table in stream mode is the more controllable design: you choose its engine, ordering, TTL and replication, and you can query it like any other table. A parse error is only one kind of bad message. A message that parses but breaks a type (a negative number for a UInt32, a string for a date) fails inside the view’s query, and that is a failure of the block, not of one message: see the partial-failure section of the materialized views guide and write the view with the tolerant ...OrZero / ...OrNull casts.

Measuring lag

A running pipeline has three places where it can fall behind: the broker (producers faster than consumers), the consumer (ClickHouse reads slowly or keeps rebalancing), and the target table (inserts slow because of merges). Lag is the newest offset in a partition minus the offset the group has committed. The broker side is the authority, so ask it:

bin/kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 \
  --describe --group clickhouse-events-v1

The output has one row per partition with CURRENT-OFFSET, LOG-END-OFFSET and LAG. Read the shape, not only the total:

Shape of lag	Likely cause
growing evenly on all partitions	the consumers are too slow overall: too few, blocks too small, target inserts slow
growing on one partition	one consumer stuck (poison message, rebalance loop) or a skewed producer key
flat but high	steady state with a backlog; the consumers match the producers and will never catch up without more capacity
no consumers listed	the group has no members: table or view detached, ClickHouse down

From the ClickHouse side, system.kafka_consumers shows what each consumer believes about itself:

SELECT `table`, consumer_id,
       assignments.topic, assignments.partition_id, assignments.current_offset,
       num_messages_read, num_commits,
       last_poll_time, last_commit_time,
       exceptions.text
FROM system.kafka_consumers
WHERE database = 'analytics';

A last_commit_time far in the past while last_poll_time is recent means the consumer polls but cannot finish inserting blocks: look at exceptions.text and at the target table (Too many parts, a readonly replica, a full disk). The consumer blocks until the view’s insert succeeds, and when the cause is fixed, consumption resumes by itself. If every poll yields a handful of messages, each becomes its own tiny part and merges cannot keep up: raise kafka_max_block_size and kafka_poll_max_batch_size, and let blocks fill with a longer kafka_flush_interval_ms. Detaching and re-attaching the queue table restarts the consumers and often clears a stuck state, but treat that as a restart, not a fix:

DETACH TABLE analytics.events_queue;
ATTACH TABLE analytics.events_queue;

Also set an alert on time lag, not only offsets: now() - max(event_time) on the target answers “how stale is the table” in the unit people care about, though a topic that is simply quiet looks as stale as one that is stuck, so read it next to the broker-side numbers.

Rebalances

A rebalance is the group coordinator reassigning partitions among consumers. It happens when a consumer joins or leaves the group (a ClickHouse restart, a DETACH/ATTACH, changing kafka_num_consumers, a new replica) and when a consumer is declared dead because it stopped heartbeating (session.timeout.ms) or stopped polling (max.poll.interval.ms). While it runs, consumption of the affected partitions pauses, and any block that was read but not yet committed is read again by whichever consumer inherits the partition: that is the mechanism behind at-least-once duplicates after a rebalance.

The classic production loop is a slow downstream insert. The consumer takes a block, the insert into the target (or a heavy chain of views) takes longer than max.poll.interval.ms, the coordinator declares the consumer dead, the partition is reassigned, the block is read again, and the cycle repeats with lag and duplicates growing. system.kafka_consumers has the evidence:

SELECT `table`, consumer_id, last_rebalance_time,
       num_rebalance_revocations, num_rebalance_assignments,
       dateDiff('second', last_commit_time, now()) AS seconds_since_commit
FROM system.kafka_consumers
WHERE database = 'analytics'
ORDER BY num_rebalance_assignments DESC;

A count that keeps climbing between runs is a loop. Fix the cause, which is nearly always time per block: make the view cheaper, make blocks smaller, speed up the target. As a last resort raise max.poll.interval.ms. Client settings of the underlying library go in the server configuration under <kafka>, with underscores in place of dots:

<clickhouse>
    <kafka>
        <max_poll_interval_ms>600000</max_poll_interval_ms>
        <auto_offset_reset>earliest</auto_offset_reset>
    </kafka>
</clickhouse>

Setting it longer hides a slow pipeline for ten minutes longer; it does not make it faster. Avoid deliberate churn during peak traffic: restarting servers one at a time with a pause in between triggers one rebalance per restart.

Retention, cleanup policy and the recovery window

Kafka deletes old data per partition, by time and by size, and whichever limit is hit first wins. A typical broker-level configuration:

log.cleanup.policy=delete
log.retention.hours=96
log.retention.bytes=1073741824
log.retention.check.interval.ms=30000

The topic-level equivalents (cleanup.policy, retention.ms, retention.bytes) override these per topic, and retention.bytes applies per partition, not per topic. The retention window is also your recovery window. If ClickHouse is down longer than the topic keeps data, the messages from the gap are gone whatever the consumer’s offset says. Size retention above the longest outage you want to survive plus the catch-up time, and alert on lag well before the window closes. Catch-up time is arithmetic: if producers write 20,000 messages/s and the pipeline can consume 60,000/s, a two-hour outage leaves a backlog of 20,000 × 7,200 = 144 million messages, drained at the 40,000/s difference in 3,600 s, one hour. If the pipeline’s headroom is only 10%, the same backlog takes twenty hours to drain, and any retention shorter than that loses data during the recovery itself.

cleanup.policy=compact is different: it keeps the latest record per key indefinitely instead of deleting by age. It suits changelog-style topics, where reading from the beginning gives the current state of every key. Pair it on the ClickHouse side with a ReplacingMergeTree keyed by that key. A compacted topic still contains the not-yet-compacted history, and delete markers are a record with a key and an empty value, so decide, and test, how your format and view treat a message with no payload before pointing the engine at such a topic.

Topic creation matters too. A producer or consumer touching a topic that does not exist may cause the broker to auto-create it with the broker default partition count (num.partitions, normally 1), which silently caps your parallelism at one consumer. Disable auto.create.topics.enable in production and create topics explicitly with the partition count you planned.

Pausing, changing and replaying

To change the view’s SELECT, or to pause consumption, detach the view first, then re-attach or recreate it:

DETACH TABLE analytics.events_mv;
-- change the schema or recreate the view here
ATTACH TABLE analytics.events_mv;

While the view is detached nothing reads the topic, the committed offsets stay where they were, and the backlog grows in Kafka. When you attach again the consumer resumes from the committed offset, as long as it has not expired and the retention window has not eaten the backlog (see above).

To replay, stop the group’s consumers (detach the table), reset the offsets with the Kafka tool, and attach again. The group must have no active members or the reset is refused.

# preview first
bin/kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 \
  --group clickhouse-events-v1 --topic events \
  --reset-offsets --to-earliest --dry-run

# apply
bin/kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 \
  --group clickhouse-events-v1 --topic events \
  --reset-offsets --to-earliest --execute

Other targets are --to-latest, --to-datetime <ISO-8601>, --shift-by <n> and --to-offset <n>. A replay produces duplicates in the target, so decide first how you will handle them: truncate and reload, deduplicate on a business id with a ReplacingMergeTree, or delete the replayed range by its kafka_partition and kafka_offset columns before you start. The coordinates are the reason to keep them in the target.

Setting kafka_keeper_path and kafka_replica_name switches to a different engine implementation that stores offsets in ClickHouse Keeper instead of in the Kafka brokers; the documentation describes it as experimental, so do not move a production pipeline there on the strength of this page.

Operations checklist

1. Topic: partition count chosen with consumer parallelism and headroom in mind; replication factor 3; explicit creation; auto.create.topics.enable off.
2. Group: versioned group name (-v1); one group per independent reader; consumers per table times servers no larger than partitions.
3. Target: explicit TO table; Kafka coordinates stored as columns; a decision, written down, on how duplicates are handled.
4. Errors: kafka_handle_error_mode set to stream (with quarantine table and TTL) or dead_letter_queue (with the server section configured); the main view filters on _error; view casts are tolerant.
5. Blocks and parts: flush interval and block size chosen from the parts-per-minute arithmetic, checked against active parts per partition in system.parts.
6. Monitoring: broker-side lag per partition; system.kafka_consumers for exceptions.text, last_commit_time age and rebalance counters; time lag on the target; alerts that fire well before the retention window.
7. Retention: topic retention longer than outage plus catch-up time, with the catch-up computed from real produce and consume rates.
8. Runbook: how to pause (detach the view), restart (detach and attach the queue table), and replay (stop, reset offsets with --dry-run first, attach), and which duplicates a replay creates.

Kafka CLI cheat sheet

The scripts live in the bin/ directory of the Kafka distribution. All of them take --bootstrap-server.

# topics
bin/kafka-topics.sh --bootstrap-server kafka-1:9092 --list
bin/kafka-topics.sh --bootstrap-server kafka-1:9092 --create --topic sample --partitions 3 --replication-factor 3
bin/kafka-topics.sh --bootstrap-server kafka-1:9092 --describe --topic sample
bin/kafka-topics.sh --bootstrap-server kafka-1:9092 --delete --topic sample

# produce and consume from a terminal
bin/kafka-console-producer.sh --bootstrap-server kafka-1:9092 --topic sample
bin/kafka-console-consumer.sh --bootstrap-server kafka-1:9092 --topic sample --from-beginning

# consumer groups
bin/kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 --list
bin/kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 --describe --group clickhouse-events-v1

# end offset of every partition
bin/kafka-get-offsets.sh --bootstrap-server kafka-1:9092 --topic sample

The console producer is the quickest way to test a pipeline: paste a JSON line matching your kafka_format and watch it arrive in the target table within the flush interval. Paste a deliberately broken line next, to prove the quarantine path works before production does it for you.

Reference: the Kafka engine page lists every setting, and system.dead_letter_queue documents the dead-letter table.