How the ClickHouse Kafka Engine Commits Offsets Only After the Block Is Written

TL;DR: A Kafka consumer that commits its offset before the data is durable loses data; one that commits after duplicates data. ClickHouse picks the second: the background task builds a block, pushes it through the materialized views, and only when the pipeline has finished does it call commit(). If the block is cancelled halfway, the consumer rewinds to the start of that block, so the block is replayed instead of lost. The cost is a real hole — the code says so in an exception message.

Kafka is a log with a cursor, and every consumer of a log has to answer the same question: when do I tell the broker I have read this? Answer it too early and a crash between the commit and the write loses data; answer it too late and a crash between the write and the commit duplicates data. Neither is “exactly once”, and the difference between them is where the correctness bug lives.

ClickHouse’s Kafka engine answers it as late as it can: the offset commit happens after the rows have been written into the materialized views. This article follows that ordering through the code, and then through the places where it leaks.

The shape of the engine

A Kafka table in ClickHouse is not a table you query — it is a stream source with materialized views attached:

sql
CREATE TABLE events_queue (id UInt64, payload String)
ENGINE = Kafka
SETTINGS kafka_broker_list = 'kafka:9092',
         kafka_topic_list = 'events',
         kafka_group_name = 'clickhouse',
         kafka_format = 'JSONEachRow';

CREATE MATERIALIZED VIEW events_mv TO events AS SELECT * FROM events_queue;

The engine’s job is to read messages, parse them with a format, and push the resulting blocks into the views. It never stores anything itself. The background work is a scheduled task, one per consumer (or one for all of them, unless kafka_thread_per_consumer is set):

cpp
// src/Storages/Kafka/StorageKafka.cpp:217
        auto task = getContext()->getMessageBrokerSchedulePool()->createTask(getStorageID(), log->name(), [this, i]{ threadFunc(i); });

threadFunc is the outer loop: it checks that the dependent views are ready, claims a “cycle” (so that two tasks do not consume the same topic concurrently), and then streams until it stalls or hits the work-duration limit. The work itself is in streamToViews, and this is where the ordering lives.

The push: write first, commit after

streamToViews builds an INSERT pipeline into the dependent views, runs it to completion, and only then commits the offsets:

cpp
// src/Storages/Kafka/StorageKafka.cpp:757
    std::atomic_size_t rows = 0;
    {
        block_io.pipeline.complete(std::move(pipe));

        // we need to read all consumers in parallel (sequential read may lead to situation
        // when some of consumers are not used, and will break some Kafka consumer invariants)
        block_io.pipeline.setNumThreads(stream_count);
        block_io.pipeline.setConcurrencyControl(kafka_context->getSettingsRef()[Setting::use_concurrency_control]);

        block_io.pipeline.setProgressCallback([&](const Progress & progress) { rows += progress.read_rows.load(); });
        CompletedPipelineExecutor executor(block_io.pipeline);
        executor.execute();
    }

    bool some_stream_is_stalled = false;
    for (auto & source : sources)
    {
        some_stream_is_stalled = some_stream_is_stalled || source->isStalled();
        source->commit();
    }

Read the two halves again. executor.execute() runs the whole pipeline: Kafka source → format parser → INSERT into the materialized views → the view’s own storage engine (usually a MergeTree, which writes a part). When it returns, the data is in the destination table. source->commit() then tells Kafka “I have read up to offset N”.

If the process dies during executor.execute(), the offsets were never committed, so the messages are re-read after restart, and the destination table does not have them. No data loss. If the process dies between executor.execute() and source->commit(), the destination table has the rows and Kafka does not know it: the messages will be re-read and re-written. Duplicates.

That window is unavoidable without a transaction spanning Kafka and ClickHouse, and the code says so out loud — the exception message when the commit fails is one of the best pieces of honest error reporting in the codebase:

cpp
// src/Storages/Kafka/KafkaConsumer.cpp:257
        if (!committed)
        {
            // TODO: insert atomicity / transactions is needed here (possibility to rollback, on 2 phase commits)
            ProfileEvents::increment(ProfileEvents::KafkaCommitFailures);
            throw Exception(ErrorCodes::CANNOT_COMMIT_OFFSET,
                            "All commit attempts failed. Last block was already written to target table(s), "
                            "but was not committed to Kafka.");
        }

“Last block was already written to target table(s), but was not committed to Kafka.” That is the exactly-once hole, named precisely. The TODO says what would close it: a two-phase commit.

Note also that the commit is retried five times before giving up, because a Kafka commit can fail for boring transient reasons — “broker may reject commit if during offsets.commit.timeout.ms (5000 by default), there were not enough replicas available for the __consumer_offsets topic”. One error code is treated as success: RD_KAFKA_RESP_ERR__NO_OFFSET means “there was nothing to commit”, so it short-circuits the retry loop. Everything else is retried, and when the retries run out:

The block boundary

The commit is per block, not per message. The size of a block is decided by the source, and the source’s loop has four exit conditions:

cpp
// src/Storages/Kafka/KafkaSource.cpp:323
        if (!consumer->hasMorePolledMessages()
            && (total_rows >= max_block_size || !checkTimeLimit() || failed_poll_attempts >= MAX_FAILED_POLL_ATTEMPTS))
        {
            break;
        }

total_rows >= max_block_size — the block is full.
!checkTimeLimit() — the flush interval elapsed. The default is stream_flush_interval_ms = 7500 ms, or kafka_flush_interval_ms if it is set.
failed_poll_attempts >= MAX_FAILED_POLL_ATTEMPTS — the topic is empty and the poll has failed 10 times in a row (“with default poll timeout (500ms) it will give about 5 sec delay for doing 10 retries when selecting from empty topic”).
!consumer->hasMorePolledMessages() — the polled batch is exhausted, so there is nothing more to read without another poll.

The block size is max_insert_block_size / num_consumers by default, so that N consumers each produce a block of 1/N of the insert size (or kafka_max_block_size if it is set explicitly). A bigger block means fewer commits and fewer parts; a smaller block means lower latency for the first rows. This is the knob.

The block start: how to replay

Here is the part I find genuinely clever. When a block is aborted — the table is detached, the query is cancelled, the view is not ready — the engine must not lose the messages it already consumed but did not commit. It must also not re-consume messages it already committed.

The trick: the consumer records where the current block started, per topic-partition, before it consumes anything:

cpp
// src/Storages/Kafka/KafkaConsumer.cpp:586
void KafkaConsumer::storeLastReadMessageOffset()
{
    if (!isStalled())
    {
        trackCurrentBlockStart(*(current - 1));
        consumer->store_offset(*(current - 1));
        ++offsets_stored;
    }
}

void KafkaConsumer::trackCurrentBlockStart(const cppkafka::Message & message)
{
    for (const auto & tp : block_start_offsets)
        if (tp.get_partition() == message.get_partition() && tp.get_topic() == message.get_topic())
            return;
    block_start_offsets.emplace_back(message.get_topic(), message.get_partition(), message.get_offset());
}

storeLastReadMessageOffset() is called after every message that produced rows. It stores the current offset (the one to commit later) and records the first offset seen for this topic-partition since the last commit — trackCurrentBlockStart is idempotent per partition, so only the first one is kept.

And when the block is cancelled, the source rewinds to those recorded starting points:

cpp
// src/Storages/Kafka/KafkaSource.cpp:330
    if (cancelled || storage.isConsumeCancelRequested(cancel_epoch))
    {
        try
        {
            consumer->rewindToLastCommitted();
        }
        catch (const cppkafka::HandleException & e)
        {
            LOG_WARNING(log, "Failed to rewind to last committed offset on abort: {}", e.what());
            consumer->cleanBlockStartOffsets();
            consumer->markDirty();
        }
        return {};
    }

and the rewind assigns the consumer back to the block’s start offsets and commits them, so the rewind is durable:

cpp
// src/Storages/Kafka/KafkaConsumer.cpp:411
    /// Rewind to where the in-flight block started. Re-commit the block partitions
    size_t max_retries = 5;
    while (max_retries > 0)
    {
        try
        {
            auto positions = consumer->get_offsets_committed(consumer->get_assignment());
            for (auto & tp : positions)
                for (const auto & start : block_start_offsets)
                    if (tp.get_partition() == start.get_partition() && tp.get_topic() == start.get_topic())
                        tp.set_offset(start.get_offset());

            consumer->assign(positions);
            consumer->commit(block_start_offsets);
            LOG_TRACE(log, "Aborted in-flight block, rewound to block start: {}", block_start_offsets);
            cleanBlockStartOffsets();
            return;
        }

So the failure taxonomy is:

text
   block in flight (offsets stored, not committed)
        |
        |  crash / cancel / view not ready
        v
   rewind to block_start_offsets  ->  the block is re-read from the broker
        |
        |  the block is re-consumed and re-written
        v
   duplicates in the destination (the previous attempt may have
   partially or fully written before the failure)

The rewind is what makes the engine not lose data. The duplicates are the price, paid in the destination table — which is why the standard advice is to use a deduplicating engine (ReplacingMergeTree, CollapsingMergeTree) or to make the INSERT idempotent. If there are no recorded block starts (because kafka_commit_every_batch committed mid-block), the rewind is a plain “go back to the last committed offset”.

The other mode: commit every batch

The default is to commit once per block, after the write. The setting kafka_commit_every_batch (default false) changes the trade-off: when it is on, the consumer commits each polled batch as it is consumed, before the block is written:

cpp
// src/Storages/Kafka/KafkaConsumer.cpp:547
    if (hasMorePolledMessages())
        return getNextMessage();

    if (intermediate_commit)
        commit();

    doPoll();

The intermediate_commit happens right before a new poll, i.e. at the boundary between batches. This is the at-most-once mode: if the server dies after the commit but before the write, the messages are gone.

So the two settings give the two ends of the classic trade-off:

setting	commit point	failure mode
`kafka_commit_every_batch = false` (default)	after the block is written	duplicates on crash
`kafka_commit_every_batch = true`	after each polled batch	loss on crash

The default picks duplicates: in a stream-processing pipeline duplicates are usually recoverable (dedup downstream) and losses are not.

Reading from a Kafka table directly

A SELECT from a Kafka table is a different path, and a strange one: the table is read destructively, consuming the messages.

cpp
// src/Storages/Kafka/StorageKafka.cpp:146
        for (size_t i = 0; i < kafka_storage.num_consumers; ++i)
        {
            /// Use block size of 1, otherwise LIMIT won't work properly as it will buffer excess messages in the last block
            /// TODO: probably that leads to awful performance.
            /// FIXME: seems that doesn't help with extra reading and committing unprocessed messages.
            pipes.emplace_back(std::make_shared<KafkaSource>(
                kafka_storage,
                storage_snapshot,
                modified_context,
                column_names,
                kafka_storage.log,
                1,
                (*kafka_storage.kafka_settings)[KafkaSetting::kafka_commit_on_select]));
        }

Note the block size of 1: with a larger block, LIMIT 5 would have to read and buffer a whole block of messages to return five rows, and the extra messages would be committed anyway. With a block size of 1 the LIMIT works, at the cost of one pipeline setup per row.

And the commit is done in the suffix of the source — KafkaSource::generate() commits only when the source is exhausted and the whole result has been consumed by the client — which is why the setting is called kafka_commit_on_select and defaults to false. With it off, a SELECT from a Kafka table consumes the messages but never commits their offsets, so the group will read them again later; with it on, the offsets are committed when the select finishes, and the messages are gone for good.

The consumer pool

One more piece of machinery that is easy to miss: the Kafka consumers are pooled. The engine creates num_consumers consumer objects up front, and a SELECT or a streaming task borrows one with popConsumer(timeout).

The pool exists because creating a cppkafka::Consumer and subscribing is expensive (it involves a group join and a rebalance). popConsumer() has three cases, in order: a free consumer exists, so return it; no consumer is free but a slot has been closed, so create a new one under the lock (cheap, because creating the client does not subscribe yet); otherwise wait up to the timeout for one to be released.

Unused consumers are closed by a cleanup thread after kafka_consumers_pool_ttl_ms (60 seconds by default), so a burst of SELECTs does not leave a hundred subscribed consumers hanging around.

The modern version: an RAII offset guard

The engine that ships alongside the old one — StorageKafka2 — solves the same problem with a different tool: a guard object whose destructor rolls the offset back.

cpp
// src/Storages/Kafka/KeeperHandlingConsumer.h:60
    class OffsetGuard
    {
    public:
        OffsetGuard(KeeperHandlingConsumer & consumer_, int64_t new_offset_);
        OffsetGuard(OffsetGuard && other) noexcept;
        ~OffsetGuard();

        void commit();
    private:
        bool needs_rollback{true};
        KeeperHandlingConsumer * consumer;
        int64_t new_offset;
    };

and its destructor and commit() are the whole mechanism:

cpp
// src/Storages/Kafka/KeeperHandlingConsumer.cpp:97
KeeperHandlingConsumer::OffsetGuard::~OffsetGuard()
{
    if (consumer && needs_rollback)
        consumer->rollbackToCommittedOffsetsNoThrow();
}

void KeeperHandlingConsumer::OffsetGuard::commit()
{
    chassert(needs_rollback && consumer);
    consumer->saveCommittedOffset(new_offset);
    needs_rollback = false;
}

The guard is created when the messages are polled and rolls the offset back unless someone explicitly calls commit() on it. The source takes the guard only after it knows it will not throw — “the cancel may have arrived while pollConsumer was polling. Drop the block by not taking its offset guard” — and the destructor of the source releases it: “if commit was not called, the OffsetGuard destructor rolls back”.

This is the same “commit after the write” ordering as the old engine, but expressed as a scope: the offset advances if and only if the block survived to the end of the scope. Every early return, every exception, every cancellation rolls it back automatically. In this version the offsets live in ClickHouse Keeper rather than in Kafka itself, so the engine can be replicated.

Trade-offs

What the ordering buys: a crash never loses a message that ClickHouse had not yet written. The failure mode is duplicates, which a downstream ReplacingMergeTree or a deduplicating materialized view can absorb.

What it costs:

There is no exactly-once. The window between “the block is in the MergeTree” and “the offset is committed” is a window in which a crash duplicates. The exception message names the missing piece: “insert atomicity / transactions is needed here”.
The commit is a synchronous round trip to the broker, in the middle of the streaming loop. A broker that cannot accept the commit (no replicas for __consumer_offsets) stalls the whole stream.
kafka_commit_every_batch = true trades duplicates for loss — the right choice when a message’s content makes an INSERT idempotent anyway, and the wrong choice when it does not.
The rewind is per-partition and per-block, and the “rewind” is itself a commit, so an aborted block costs an extra broker round trip.
A SELECT from a Kafka table is destructive and, with a block size of 1, slow. It is a debugging tool, not a query interface.

Wrapping up

The one thing to remember: the commit is after the write — a deliberate choice of “duplicates on crash” over “data loss on crash”, enforced by the structure of the code (executor.execute() then source->commit()) rather than by a comment.

The part people get wrong: a crash between the write and the commit duplicates the block, and nothing in ClickHouse cleans that up for you. If your pipeline cannot tolerate duplicates, the destination engine has to be a deduplicating one.

What to go do about it: watch KafkaCommitFailures in system.events. It should be zero. If it is not, the error message in the server log is the exact sentence quoted above.