ClickHouse Kafka Engine: Secure Setup and 5 Silent Failures

Tested in November 2026 on ClickHouse 26.8 LTS and Kafka 3.9.1 (AutoMQ 1.7.4), AWS us-east-1 ·

A ClickHouse Kafka engine table can be created without a single error and still read 0 rows. In our tests, five different mistakes did exactly that.

The ClickHouse Kafka engine streams Kafka to ClickHouse with three objects. A Kafka engine table consumes the topic, a MergeTree table stores the rows, and a materialized view moves each batch between them. On ClickHouse 26.8, SASL_SSL and SCRAM are four ordinary table settings, with no config.xml change.

This tutorial is for any ClickHouse not on ClickHouse Cloud. You’ll get a secured pipeline, a troubleshooting guide to the five silent failures, a dedup key, and batch settings measured on S3-backed storage. Every command, error and number comes from real runs on 5 November 2026.

Key Findings

• No extra configuration needed: On ClickHouse 26.8, you can configure SASL_SSL and SCRAM directly in the Kafka table settings. There is no need to modify config.xml.
• Wrong connection settings may go unnoticed: ClickHouse still creates the table even if the SASL mechanism, SCRAM version, password or security protocol is wrong. However, no messages reach the table. You can find the connection error in system.kafka_consumers.
• One bad JSON message can stop ingestion: In the default error-handling mode, a single malformed JSON message can stall the entire Kafka ingestion pipeline. In our test on ClickHouse 26.8, ClickHouse attempted to read the same batch about 41 times, but no rows were inserted.
• No data loss or duplicates in our interruption tests: Across six clean interruptions on ClickHouse 26.8, we tested 450,000 Kafka messages. All messages arrived, with zero duplicates and zero lost rows. However, Kafka ingestion is still an at-least-once process, so this does not guarantee exactly-once delivery in every failure scenario.
• Longer flush intervals reduce storage overhead: On ClickHouse 26.8 at 1,000 messages per second, increasing kafka_flush_interval_ms from its default of 7.5 seconds to 60 seconds (with kafka_max_block_size = 100000) reduced the number of new data parts from 7.7 to 1.3 per minute. S3 write requests per 3-minute run also dropped from 315 to 44.The trade-off was higher ingestion latency: p99 delay increased from 10.6 to 61.2 seconds.

Every test in this post ran on managed ClickHouse and Kafka from Selfhost.dev, in one AWS region. See the tested ClickHouse instance (t4g.large, us-east-1, S3-backed) priced, no signup

How do I connect Kafka to ClickHouse?

To connect Kafka to ClickHouse, create three objects in SQL. Consumption starts when the view is attached and stops when it is detached.

• A Kafka engine table consumes the topic. It holds the broker address, topic, consumer group and credentials, and stores nothing.
• A materialized view moves each batch from the Kafka table to the MergeTree table. This is where you rename, cast or drop fields.
• A MergeTree table stores the rows you query.

producer (Python, confluent-kafka)
    │   SASL_SSL + SCRAM-SHA-256, port 9092
    ▼
Kafka topic "events" (3 partitions)
    │   consumer group "clickhouse_events"
    ▼
events_queue    ENGINE = Kafka          consumer, stores nothing
    │   one block per flush
    ▼
events_mv       MATERIALIZED VIEW       SELECT ... FROM events_queue
    │   INSERT
    ▼
events          ENGINE = MergeTree      rows live here

ClickHouse Kafka integration: Kafka engine, Connect sink or ClickPipes?

A ClickHouse Kafka integration runs in one of three places. The Kafka engine runs inside ClickHouse, the ClickHouse Kafka connector runs on your Kafka Connect workers, and ClickPipes is managed, on ClickHouse Cloud only.

Guarantees are as ClickHouse documents them for the Kafka Connect sink and ClickPipes. Vector or a custom consumer also work.

• Use the Kafka engine when your ClickHouse isn’t on ClickHouse Cloud and you want ingestion as SQL, with no extra service to run.
• Use the ClickHouse Kafka Connect sink when you already run Kafka Connect, or need exactly-once inserts.
• Use ClickPipes when your ClickHouse runs on ClickHouse Cloud. Here’s how ClickHouse Cloud compares with other managed ClickHouse services.

What you need before you start

Before you connect Kafka to ClickHouse, you need three things: the Kafka connection details, network access from ClickHouse to Kafka port 9092, and a ClickHouse SQL user that can create tables and views.

1. Copy the Kafka connection details. The bootstrap server, Kafka user and password are on the Kafka instance’s Connection tab in Selfhost.dev managed Kafka.

1. Allow ClickHouse’s private IP on Kafka port 9092. Inside the VPC, the Kafka hostname resolves to the broker’s private IP. So ClickHouse connects from its private address, not its public one.

• The console doesn’t show a database’s private IP. Read it from the API: selfhost api /aws/v1/instances/<clickhouse-pid> returns it as private_ip.
• On the Kafka instance, open the Configuration tab, choose Network access, add <clickhouse-private-ip>/32 under Add IP range, and select Save changes.
• ClickHouse reached Kafka only after we did this.

• 3. Use a ClickHouse SQL user that can create tables and views. We used the instance’s SQL user, chadmin.

How we tested

• ClickHouse 26.8.36.3 and Kafka 3.9.1 on AutoMQ 1.7.4, each on a t4g.large (2 vCPU, 8 GB) in us-east-1. In our test, both instances were in the same VPC. t4g instances are burstable, so sustained throughput depends on CPU credits.
• Kafka listener: SASL_SSL with SCRAM-SHA-256 on port 9092, with a Let’s Encrypt certificate.
• Topics with 3 partitions and replication factor 1. On AutoMQ, topic data lives in S3, so losing a broker doesn’t lose acknowledged messages.
• Messages: keyed JSON, 256 bytes each.
• MergeTree parts on an S3-backed disk, object_storage_cache.
• Producer: Python confluent-kafka on a laptop in India, which tops out around 1,400 msg/s. For large backlogs we produced from ClickHouse itself: an INSERT INTO a Kafka engine table wrote 3,000,000 messages in about 15 s.

To build this same pipeline, create a Kafka and a ClickHouse instance in the same region, then run the SQL below as written. On Selfhost.dev both come from one console, and new accounts get $5 of credit with no card. Create your Kafka and ClickHouse instances

How do you set up ClickHouse Kafka SSL (SASL_SSL with SCRAM)?

Put four security settings in the Kafka table’s SETTINGS list: kafka_security_protocol = ‘sasl_ssl’, kafka_sasl_mechanism, kafka_sasl_username and kafka_sasl_password. On ClickHouse 26.8 these are ordinary table settings, so config.xml stays untouched.

This is the full ClickHouse Kafka engine example we ran. Replace the placeholders with values from the Connection tab:

CREATE TABLE default.events_queue
(
    id      UInt64,
    run     UInt8,
    user_id UInt32,
    event   String,
    ts      Float64,
    pad     String
)
ENGINE = Kafka
SETTINGS
    kafka_broker_list       = '<bootstrap-host>:9092',
    kafka_topic_list        = 'events',
    kafka_group_name        = 'clickhouse_events',
    kafka_format            = 'JSONEachRow',
    kafka_security_protocol = 'sasl_ssl',
    kafka_sasl_mechanism    = 'SCRAM-SHA-256',
    kafka_sasl_username     = '<kafka-user>',
    kafka_sasl_password     = '<kafka-password>',
    kafka_num_consumers     = 1;

We didn’t need a CA file because the broker’s certificate is from Let’s Encrypt. Use the security protocol the Connection tab shows: a new Selfhost.dev Kafka instance uses SASL_PLAINTEXT for a few minutes, until its TLS certificate lands.

What each Kafka engine setting does

A new consumer group starts from the earliest offset. ClickHouse sets auto.offset.reset to earliest (KafkaConfigLoader.cpp in the ClickHouse source on GitHub), and each fresh group in our tests read its topic from the start. Keep the group name when you recreate the table, and it resumes from the committed offsets.

Create the MergeTree table and the Kafka engine materialized view

The Kafka engine materialized view reads each batch from the Kafka table and inserts it into a MergeTree table. Create the target table first: ours keeps the fields we query, three virtual columns and an insert timestamp.

CREATE TABLE default.events

(
    id            UInt64,
    user_id       UInt32,
    event         String,
    ts            Float64,
    _partition    UInt64,
    _offset       UInt64,
    _timestamp_ms Nullable(DateTime64(3)),
    inserted_at   DateTime64(3) DEFAULT now64(3)
)
ENGINE = MergeTree
ORDER BY (user_id, id);

ORDER BY (user_id, id) sorts each part for per-user queries. It is not a uniqueness constraint: a plain MergeTree keeps every duplicate, so we also store _partition and _offset.

Then the view, which starts consuming as soon as it exists:

CREATE MATERIALIZED VIEW default.events_mv TO default.events AS
SELECT id, user_id, event, ts, _partition, _offset, _timestamp_ms
FROM default.events_queue;

The view drops run and pad. inserted_at takes its default on insert; the Kafka table itself can’t have DEFAULT columns.

How to check that every Kafka message arrived

Produce a known number of messages, then compare three counts: what the producer reported, count(), and the distinct (_partition, _offset) pairs. This is the producer we used:

# /// script
# dependencies = ["confluent-kafka"]
# ///
"""Produce keyed, fixed-size JSON messages at a fixed rate to a SASL_SSL/SCRAM Kafka.
Usage: produce.py --topic T --rate 1000 --seconds 600 [--size 256] [--run R] [--bad-every 0]
Env: KAFKA_BOOTSTRAP, KAFKA_USER,
import argparse, json, os, time
from confluent_kafka import Produ
from confluent_kafka.admin import AdminClient, NewTopic

ap = argparse.ArgumentParser()
ap.add_argument("--topic", defaul
ap.add_argument("--rate", type=int, default=1000)
ap.add_argument("--seconds", type
ap.add_argument("--size", type=int, default=256, help="bytes per JSON message")
ap.add_argument("--run", type=int
ap.add_argument("--bad-every", type=int, default=0, help="send a malformed message every N")
a = ap.parse_args()

conf = {
    "bootstrap.servers": os.environ["KAFKA_BOOTSTRAP"],
    "security.protocol": "SASL_SS
    "sasl.mechanism": "SCRAM-SHA-256",
    "sasl.username": os.environ["
    "sasl.password": os.environ["KAFKA_PASS"],
}
admin = AdminClient(conf)
if a.topic not in admin.list_topi
    for f in admin.create_topics([NewTopic(a.topic, 3, 1)]).values():
        f.result()
    print(f"created topic {a.topic} (3 partitions, RF 1)", flush=True)

p = Producer({**conf, "linger.ms": 5, "queue.buffering.max.messages": 500000})

def msg(i):
    base = {"id": i, "run": a.runt": "click", "ts": time.time(),"pad": ""}
    pad = a.size - len(json.dumps
    base["pad"] = "x" * max(pad, 0)
    return json.dumps(base)

total = a.rate * a.seconds
start = time.monotonic()
sent = 0
for i in range(total):
    if a.bad_every and i % a.bad_
        value = '{"id": ' + str(i) + ', "run": broken'
    else:
        value = msg(i)
    p.produce(a.topic, key=str(i)
    sent += 1
    p.poll(0)
    target = start + sent / a.rate
    delay = target - time.monoton
    if delay > 0:
        time.sleep(delay)
    if sent % (a.rate * 10) == 0:
        print(f"{time.strftime('%tal}", flush=True)
left = p.flush(60)
el = time.monotonic() - start
print(f"produced {sent - left}/{total} to {a.topic} in {el:.1f}s ({(sent-left)/el:.0f} msg/s),
size {len(msg(0))} B", flush=True

For the pipeline above:

KAFKA_BOOTSTRAP='<bootstrap-host>:9092' KAFKA_USER='<kafka-user>' KAFKA_PASS='<kafka-password>' \
  uv run produce.py --topic events --rate 1000 --seconds 60

Then compare in ClickHouse:

SELECT count() AS rows, uniqExact(_partition, _offset) AS unique_messages
FROM default.events;

In run 1 of the duplicate test, the producer reported 60,000 of 60,000 messages produced, and the check returned:

rows    unique_messages
60000   60000

All three counts match: nothing lost, nothing doubled. Run this check after every change below.

ClickHouse Kafka engine virtual columns: _timestamp, _offset and _partition

Every ClickHouse Kafka engine table exposes message metadata as virtual columns such as _partition, _offset and _timestamp_ms. They exist only on the engine table, so the materialized view must select them into real columns. Store these five:

Our test tables stored _partition, _offset and _timestamp_ms. Add _topic and _timestamp the same way. _key, _headers.name and _headers.value are also available.

Measure end-to-end delay as insert time minus message time. inserted_at holds now64(3) at the moment the view inserts the block:

SELECT
    round(quantile(0.5)(dateDiff('millisecond', _timestamp_ms, inserted_at)) / 1000, 1)  AS p50_delay_s,
    round(quantile(0.99)(dateDiff('millisecond', _timestamp_ms, inserted_at)) / 1000, 1) AS p99_delay_s
FROM default.events;

With default settings at 1,000 msg/s:

p50_delay_s p99_delay_s
4.9 10.6

With Kafka’s default CreateTime, the producer sets the timestamp. So this delay includes the hop from our laptop in India and any clock skew. Most of it is batching: the delay tracks the flush interval (see too many parts).

Why does the Kafka engine read 0 rows? 5 silent failures

A ClickHouse Kafka engine table reads 0 rows with no visible error when its security settings are wrong or one message can’t be parsed. CREATE TABLE still succeeds. Nothing fails until the materialized view starts the consumer, and the error appears only in system.kafka_consumers:

SELECT table, exceptions.text[-1] AS last_exception
FROM system.kafka_consumers
WHERE database = 'default';

ClickHouse talks to Kafka through librdkafka, the Kafka client library built into it, so the errors use librdkafka’s wording and states (AUTH_HANDSHAKE, AUTH_REQ). We broke one setting at a time.

1. Wrong mechanism, kafka_sasl_mechanism = 'PLAIN':

SASL PLAIN mechanism handshake failed: Broker: Unsupported SASL mechanism: broker's supported mechanisms: SCRAM-SHA-256 (after 186ms in state AUTH_HANDSHAKE)

2. Wrong SCRAM variant, kafka_sasl_mechanism = ‘SCRAM-SHA-512’:

SASL SCRAM-SHA-512 mechanism handshake failed: Broker: Unsupported SASL mechanism: broker's supported mechanisms: SCRAM-SHA-256 (after 200ms in state AUTH_HANDSHAKE)

Fix for both: SCRAM-SHA-256, the mechanism the broker names.

3. Wrong password, kafka_sasl_password = 'wrong-password-123':

SASL authentication error: Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-256 (after 229ms in state AUTH_REQ)

Fix: copy the password from the Connection tab again.

4. Wrong protocol, kafka_security_protocol = 'sasl_plaintext' against the TLS listener. This one is hard to spot. system.kafka_consumers only says:

1/1 brokers are down

The useful hint is in system.text_log:

SELECT event_time, message
FROM system.text_log
WHERE message LIKE '%APIVERSION_QUERY%'
ORDER BY event_time DESC
LIMIT 5;

Disconnected: connection closed by peer: receive 0 after POLLIN (after 2ms in state APIVERSION_QUERY)

The client sent a plaintext request to a TLS port, and the broker closed the connection 2 ms in. Fix: sasl_ssl. If you see “brokers are down” and the network path is open, search text_log for APIVERSION_QUERY first.

5. One malformed message. In the default error mode, one bad JSON message makes the consumer reread the same batch forever, and 0 rows land. The fix, stream mode, is in the next section.

How does the Kafka engine handle bad messages? (kafka_handle_error_mode)

By default, it doesn’t: one malformed message stops the whole pipeline. Set kafka_handle_error_mode = 'stream' and each broken message becomes a row with its raw text in _raw_message and the parser error in _error.

We produced 100 messages to topic bad, with every 10th one malformed:

{"id": 9, "run": broken

Default mode. Zero rows landed. The first bad message fails the whole batch, the offset isn’t committed, and the consumer reads the same batch again, forever. When we checked, num_messages_read was 4,100 for a 100-message topic: the same batch had been read about 41 times. The error, from system.kafka_consumers.exceptions:

Code: 27. DB::Exception: Cannot parse input: expected ',' before: 'broken': (at row 1) : while parsing Kafka message (topic: bad, partition: 0, offset: 1): While executing Kafka. (CANNOT_PARSE_INPUT_ASSERTION_FAILED)

Stream mode. A broken message becomes a row with default values, plus _raw_message and _error. Store both:

CREATE TABLE default.bad_queue
(
    id UInt64, run UInt8, user_id UInt32, event String, ts Float64, pad String
)
ENGINE = Kafka
SETTINGS
    kafka_broker_list       = '<bootstrap-host>:9092',
    kafka_topic_list        = 'bad',
    kafka_group_name        = 'clickhouse_bad',
    kafka_format            = 'JSONEachRow',
    kafka_security_protocol = 'sasl_ssl',
    kafka_sasl_mechanism    = 'SCRAM-SHA-256',
    kafka_sasl_username     = '<kafka-user>',
    kafka_sasl_password     = '<kafka-password>',
    kafka_handle_error_mode = 'stream';

CREATE TABLE default.bad_stream
(
    id UInt64, user_id UInt32, event String,
    _partition UInt64, _offset UInt64,
    _raw_message String, _error String
)
ENGINE = MergeTree
ORDER BY (_partition, _offset);

CREATE MATERIALIZED VIEW default.bad_stream_mv TO default.bad_stream AS
SELECT id, user_id, event, _partition, _offset, _raw_message, _error
FROM default.bad_queue;

Result: 90 good rows and 10 broken ones, each with its raw message and error. To list the broken messages:

SELECT _partition, _offset, _raw_message, _error
FROM bad_stream
WHERE length(_error) > 0;

In production, use two materialized views on the same Kafka table: – WHERE length(_error) = 0 into the main table. – WHERE length(_error) > 0 into an errors table.

kafka_skip_broken_messages also exists, but it skips bad messages instead of keeping them. We didn’t test it.

Dead letter queue. kafka_handle_error_mode = 'dead_letter_queue' writes failures to system.dead_letter_queue. The mode exists on 26.8, but the server must configure that table. On Selfhost.dev it isn’t configured, so CREATE TABLE is refused:

Code: 36. DB::Exception: The table system.dead_letter_queue is not configured on the server. You cannot create a table with this kafka_handle_error_mode. (BAD_ARGUMENTS)

Use stream mode there.

How do you monitor the Kafka engine? system.kafka_consumers and consumer lag

To monitor the ClickHouse Kafka engine, check system.kafka_consumers in ClickHouse and consumer-group lag on the Kafka broker. The first has one row per consumer, with its assignments, message counters and recent errors. Broker-side lag shows whether ClickHouse keeps up.

SELECT
    table,
    consumer_id,
    assignments.partition_id AS partitions,
    num_messages_read,
    num_commits,
    last_poll_time,
    exceptions.text[-1] AS last_exception
FROM system.kafka_consumers
WHERE database = 'default'
FORMAT Vertical;

A failing consumer, from our wrong-password test (trimmed to the columns we recorded):

Row 1:
──────
table:             events_queue
num_messages_read: 0
last_exception:    SASL authentication error: Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-256 (after 229ms in state AUTH_REQ)

exceptions.text holds the most recent errors, oldest first, so [-1] is the newest. Counters that stop moving plus a fresh exception mean a stalled consumer.

On the broker side, the Consumers tab of a managed Kafka instance on Selfhost.dev has a Consumer lag panel. It lists each consumer group’s lag in messages, most behind first.

# committed.py <topic> <group>: committed offset, end offset and lag per partition
import os, sys
from confluent_kafka import Consumer, TopicPartition

conf = {
    "bootstrap.servers": os.environ["KAFKA_BOOTSTRAP"],
    "security.protocol": "SASL_SSL",
    "sasl.mechanism": "SCRAM-SHA-256",
    "sasl.username": os.environ["KAFKA_USER"],
    "sasl.password": os.environ["KAFKA_PASS"],
    "group.id": sys.argv[2],
    "enable.auto.commit": False,
}
c = Consumer(conf); topic = sys.argv[1]
parts = c.committed([TopicPartition(topic, p) for p in range(3)], timeout=15)
for tp in parts:
    lo, hi = c.get_watermark_offsets(tp, timeout=15)
    print(f"  partition {tp.partition}: committed={tp.offset} end={hi} lag={hi - max(tp.offset, 0)}")
c.close()

Rows inserted and merges per minute come from system.part_log:

SELECT
    toStartOfMinute(event_time) AS minute,
    countIf(event_type = 'NewPart') AS new_parts,
    sumIf(rows, event_type = 'NewPart') AS rows_inserted,
    countIf(event_type = 'MergeParts') AS merges
FROM system.part_log
WHERE database = 'default' AND table = 'events'
  AND event_time > now() - INTERVAL 1 HOUR
GROUP BY minute
ORDER BY minute;

The three numbers to alert on

These thresholds are suggestions derived from our measurements. Set yours from a baseline run of your own traffic.

1. Consumer exceptions: any new entry. Every security mistake and the default-mode parse error showed up in exceptions.text while 0 rows arrived.
2. p99 end-to-end delay: above twice your measured p99. We measured 10.6 s at defaults, so alert above 21.2 s. At a 30 s interval we measured 31.6 s, so alert above 63.2 s. Run the delay query over recent rows.
3. New parts per minute: above twice your baseline. We measured 7.7 per minute at defaults, so alert above 15.4. A jump means smaller or more frequent batches, for example from more consumers or a shorter interval.

Does the ClickHouse Kafka engine create duplicates?

It can, rarely. The Kafka engine commits offsets after the insert, so a failure in between replays a batch. We forced six interruptions and measured 0 duplicates, but dedup anyway.

Each run produced at 1,000 msg/s and interrupted consumption 30 s in. Then we compared rows with distinct (_partition, _offset) pairs:

SELECT
    count() AS rows,
    uniqExact(_partition, _offset) AS unique_messages,
    rows - unique_messages AS duplicates
FROM default.events;

We measured 0 duplicates out of 450,000 messages across 6 interruptions (3 detach/attach, 3 reboots), and 0 lost rows.

Two caveats: – The reboot was a clean EC2 reboot, so ClickHouse shut down normally. We didn’t test kill -9 or power loss. - Zero in six runs isn’t exactly-once. A dropped broker connection or a rebalance between insert and commit still replays a batch. Plan for it.

How to remove duplicates: ReplacingMergeTree on (_partition, _offset)

Put _partition and _offset in the sorting key of a ReplacingMergeTree. Copies of one message then share a key, and merges keep one copy. You can keep your query columns first: copies of a message also share user_id, so the key still matches.

CREATE TABLE default.events_dedup
(
    id            UInt64,
    user_id       UInt32,
    event         String,
    ts            Float64,
    _partition    UInt64,
    _offset       UInt64,
    _timestamp_ms Nullable(DateTime64(3)),
    inserted_at   DateTime64(3) DEFAULT now64(3)
)
ENGINE = ReplacingMergeTree(inserted_at)
ORDER BY (_partition, _offset);

inserted_at as the version keeps the latest copy. Add _topic before _partition if the table reads more than one topic.

Merges run on ClickHouse’s schedule, and until one runs, a query needs FINAL to see one row per message.

To measure its cost, we loaded the 450,000 rows plus 4,640 duplicates into a ReplacingMergeTree with 3 active parts. Times are the median of 5 runs:

ClickHouse answers a plain count() from part metadata, which is why that gap is the widest. – Use FINAL where counts must be exact, such as reconciliation. – Skip it on dashboards that can tolerate a rare duplicate.

Why replication isn’t a dedup key

A ReplicatedMergeTree target is often suggested for Kafka duplicates, but replication isn’t designed as a per-message dedup key. ClickHouse’s insert deduplication works on whole blocks. Each block gets a hash of its data, and a block with a hash it has already seen is skipped (ClickHouse docs).

ClickHouse’s own Kafka table engine guide notes that this relies on identical blocks, and that events such as a Kafka rebalance can break that. A key built from the message itself, (_partition, _offset), doesn’t depend on how a batch is cut.

Can the Kafka engine lose rows?

Yes, in one rare case, and without an error, as the “Data durability” section of the Kafka engine reference explains. The offset is committed once the batch reaches the materialized views, but the new part is durable only after an fsync, which by default isn’t synchronous. If the OS page cache is lost in between, those rows vanish with no error. – A plain process kill doesn’t expose this, because the kernel still writes the page cache back. – A power loss or an unclean host reset does. Our clean reboots can’t rule it out.

The documented mitigation is two MergeTree settings on every table the batch is inserted into, cascaded view targets included:

CREATE TABLE default.events ( ... )
ENGINE = MergeTree
ORDER BY (user_id, id)
SETTINGS fsync_after_insert = 1, fsync_part_directory = 1;

With them, parts are durable before the offset is committed, which narrows the window a lot. Two limits: – It doesn’t help when the offset is committed before the insert finishes (kafka_commit_every_batch = 1, or a direct INSERT ... SELECT with kafka_commit_on_select = 1). – Asynchronous targets such as Distributed need their own settings.

Would we turn them on? Yes, for tables where a silent gap is worse than slower inserts. The fsync happens per part, not per row, so larger batches (next section) keep the count down. We did not measure the insert cost. Our parts also sat on an S3-backed disk, and we didn’t test how the page-cache window applies there.

How do you fix too many parts from the Kafka engine?

Make each batch bigger. Every flush becomes at least one new part, the unit of data ClickHouse writes and later merges. Small, frequent parts mean more merges and more S3 writes.

Two settings shape the batch: kafka_flush_interval_ms (default: stream_flush_interval_ms, 7,500 ms here) and kafka_max_block_size (rows per block).

kafka_flush_interval_ms and kafka_max_block_size, measured

Each configuration ran at 1,000 msg/s for 3 minutes with one consumer, in a fresh table and group:

On ClickHouse 26.8 at 1,000 messages per second, raising the Kafka engine’s flush interval from the default 7.5 s to 60 s (with kafka_max_block_size = 100000) cut new parts from 7.7 to 1.3 per minute and S3 write requests from 315 to 44, while p99 end-to-end delay rose from 10.6 s to 61.2 s.

At this rate, the interval closes each batch, not the block size: 45,000 rows per part stayed under the 100,000 cap. – Raise kafka_max_block_size only when your rate fills a block before the interval ends. – Choose the interval from the delay you can accept. In both tuned runs, p99 landed just above it.

Both settings go in the Kafka table’s SETTINGS list:

kafka_flush_interval_ms = 60000,
kafka_max_block_size    = 100000

How we counted. We counted parts from system.part_log, not system.parts, because merges shrink the active part count. The customer SQL user can’t run SYSTEM STOP MERGES to hold them. It can’t run SYSTEM FLUSH LOGS either, so wait a few seconds for part_log to flush on its own:

SELECT
    countIf(event_type = 'NewPart') AS new_parts,
    round(new_parts / (dateDiff('second',
        minIf(event_time, event_type = 'NewPart'),
        maxIf(event_time, event_type = 'NewPart')) / 60), 1) AS parts_per_min,
    round(avgIf(rows, event_type = 'NewPart')) AS avg_rows_per_part,
    formatReadableSize(avgIf(size_in_bytes, event_type = 'NewPart')) AS avg_part_size,
    countIf(event_type = 'MergeParts') AS merges
FROM system.part_log
WHERE database = 'default' AND table = 'events';

“S3 write requests” is the change in S3PutObject plus the multipart-upload events in system.events, read before and after each run. Merges are included.

SELECT sumIf(value, event IN ('S3PutObject', 'S3CreateMultipartUpload',
                              'S3UploadPart', 'S3CompleteMultipartUpload')) AS s3_writes
FROM system.events;

kafka_num_consumers and kafka_thread_per_consumer, measured

For throughput, we drained a backlog of 3,000,000 messages of 256 B from 3 partitions. Each run used a fresh group and the default flush interval:

With kafka_thread_per_consumer = 0, the default, rows from all consumers are squashed into one block. Setting it to 1 gives each consumer its own thread and its own flush, and that was the largest single gain here.

The ClickHouse docs set two ceilings: no more consumers than partitions, and no more than the server’s physical cores. Our box had 2 vCPU, so three consumers already broke the second rule. Throughput still rose, but treat those rows as a measurement, not a sizing rule.

Do small parts cost more on S3-backed storage?

Yes: on object storage, every small part turns into S3 write requests as well as merge work. At defaults the average new part was only 70.6 KiB, so the 7.5 s default interval is the first setting to revisit.

On our instance, new MergeTree tables landed on the S3-backed object_storage_cache disk with no storage policy in the DDL. Check where yours live:

SELECT disk_name, count() AS parts, formatReadableSize(sum(bytes_on_disk)) AS size
FROM system.parts
WHERE database = 'default' AND table = 'events' AND active
GROUP BY disk_name;

On Selfhost.dev you pick the storage per ClickHouse instance: S3-backed by default, EBS only, or tiered (hot data on EBS, older data aged into S3). Compare EBS, S3 and tiered storage for ClickHouse

How do you pause and resume the Kafka engine?

Detach the materialized view, then attach it again. The Kafka table and the group’s committed offsets stay, so the consumer resumes where it stopped. Do this before changing the view’s SELECT or the target table.

DETACH TABLE default.events_mv;
-- change the view or the target table, or just wait
ATTACH TABLE default.events_mv;

We consumed 1,000 messages, detached the view, produced 3,000 more, and attached it again. Offsets come from committed.py:

• Partitions 0 and 2 were empty before the pause, so the group had no offset there yet.
• While detached, lag grew by exactly the 3,000 new messages.
• Within 20 s of ATTACH, the consumer had resumed from the committed offsets and lag was back to zero.
• The table held 4,000 rows and 4,000 distinct (_partition, _offset) pairs: nothing lost, nothing duplicated.

Detaching the Kafka table works too; runs 1 to 3 of the duplicate test did that. One limit: if a pause outlasts the broker’s offsets.retention.minutes, the group’s offsets expire. The consumer then starts from the earliest retained message, and you get duplicates.

Production checklist for the ClickHouse Kafka engine

These eight checks cover everything that broke or cost the most in our tests: security, network access, duplicates, batching, consumers, bad messages, alerts and pauses.

• Use sasl_ssl with SCRAM-SHA-256, then check system.kafka_consumers once the view attaches.
• Allow ClickHouse’s private IP as a CIDR on Kafka port 9092 before you create the Kafka table.
• Dedup with a ReplacingMergeTree keyed on _partition and _offset (plus _topic for several topics); use FINAL where counts must be exact.
• Set kafka_flush_interval_ms to the delay you can accept; 30 s cut new parts from 7.7 to 2.0 per minute.
• Keep kafka_num_consumers at or below the partition count and the core count, with kafka_thread_per_consumer = 1 when you add consumers.
• Use kafka_handle_error_mode = 'stream' and route rows with a non-empty _error to their own table.
• Alert on new consumer exceptions, and on p99 delay or new parts per minute above twice your baseline.
• Detach the materialized view before changing conversion logic, and keep the pause shorter than the broker’s offset retention.

The short version

Off ClickHouse Cloud, the ClickHouse Kafka engine is the simplest way to stream Kafka to ClickHouse: three SQL objects, no extra service, and SASL_SSL in four table settings. Its failures are quiet. Check system.kafka_consumers after every change, and confirm rows arrive before you walk away.

Selfhost.dev runs both ends of this pipeline on AWS: ClickHouse 26.8, with S3-backed, EBS or tiered storage, and Kafka-compatible brokers with a Schema Registry. You create both in the same region from one console, on one bill, billed by the hour. New accounts get $5 of credit with no card. Start a ClickHouse and Kafka pipeline

Keep reading: – Managed database services that run ClickHouse and Kafka on one account – Managed ClickHouse on AWS, with S3-backed storage – Five ways to move data from Postgres into ClickHouse – Kafka alternatives that keep your Kafka code

Frequently Asked Questions