Analytics PipelineClickHouse & Materialized Views
Guide Real-Time Data Ingestion Pipeline Implementation

Preventing Data Loss on Buffer Flush Failure

A Buffer table holds its accumulated rows entirely in RAM until a flush writes them to the destination MergeTree, which means an unclean restart, an OOM kill, or a flush that keeps failing silently discards every row in that in-memory window. The buffer offers throughput, not durability, and treating it as durable is how a pipeline loses an hour of events without a single error in the query log. This guide quantifies that exposure window and closes it with external offset tracking, a bounded staleness cap, replay on restart, and a circuit breaker that stops writers before the ceiling blocks them.

Durability has to live outside the buffer. The pattern is at-least-once delivery gated on flush confirmation: an upstream source commits its position (a Kafka offset, a Redis high-water mark) only after the rows it produced are known to be on disk in the destination, so a crash replays the uncommitted tail instead of losing it. Everything below builds that guarantee around a buffer that itself remains deliberately volatile.

Prerequisites

The exposure window

Every row lives through a window where it has been acknowledged to the writer but not yet persisted: it sits in a buffer layer between the insert’s HTTP 200 and the layer’s flush. A crash anywhere in that window loses those rows. The width of the window is bounded by max_time and by how full the layers are — which is precisely why durability depends on when you commit the upstream offset, not on the buffer at all.

The Buffer exposure window and offset-commit timing that closes it Timeline: source reads offset N, writer inserts and gets an ack, the row sits resident in a layer, the layer flushes, and only then is the row durable. The insert-ack-to-flush span is the exposure window where a crash loses RAM-resident rows. Committing the upstream offset only after flush confirmation replays the uncommitted tail on restart; committing at ack time, before flush, is unsafe. read offset N from source insert → ack HTTP 200 layer flush to MergeTree durable on disk time → EXPOSURE WINDOW · RAM only crash here loses these rows — bounded by max_time ✓ commit offset N after flush → safe replay ✗ commit at ack → tail lost on crash durability lives in when the offset is committed, not in the buffer

Step 1 — Quantify the current exposure

You cannot bound a window you have not measured. Read how many rows are resident and how long they have been waiting; multiply by your insert rate to see how much a crash would cost right now.

sql
SELECT metric, value
FROM system.asynchronous_metrics
WHERE metric LIKE '%Buffer%'
ORDER BY metric;

...BufferRows (or the buffer-specific metrics your build exposes) is the count that would vanish on an unclean stop. Compare buffer-visible rows against what is actually on disk to see the in-memory delta directly:

sql
SELECT
    (SELECT count() FROM analytics.events_buffer) AS via_buffer,
    (SELECT count() FROM analytics.events_raw)    AS on_disk,
    (SELECT count() FROM analytics.events_buffer)
      - (SELECT count() FROM analytics.events_raw) AS at_risk;

at_risk is the exact number of rows a crash would lose at this instant. Track it over time — a growing at_risk under steady load means flushes are falling behind.

Step 2 — Gate offset commits on flush confirmation

The core fix: never advance the upstream position until the rows behind it are on disk. After inserting a batch, force the layers holding it to drain, then commit the offset. OPTIMIZE TABLE on a buffer flushes synchronously and returns only when the destination has the rows.

python
import clickhouse_connect

client = clickhouse_connect.get_client(host="clickhouse", database="analytics")

def insert_and_commit(rows, columns, kafka_consumer, batch):
    # 1. Write to the volatile buffer.
    client.insert("events_buffer", rows, column_names=columns)
    # 2. Force the buffer to persist before we advance the source position.
    client.command("OPTIMIZE TABLE analytics.events_buffer")
    # 3. Only now is it safe to commit — a crash before this replays the batch.
    kafka_consumer.commit(message=batch[-1])

Draining on every batch trades some of the buffer’s coalescing benefit for durability. In high-throughput pipelines, commit on a cadence instead: flush and commit every few seconds so at most one interval’s worth of offsets is ever uncommitted. The consumer-group offset mechanics for the Kafka case are covered in configuring Kafka consumer groups for ClickHouse.

Step 3 — Cap staleness with max_time

If you would rather not force-drain per batch, bound the exposure window structurally by shrinking max_time. A layer then flushes on its own within that ceiling regardless of row count, so the worst-case loss is one max_time interval of ingest.

sql
-- Recreate with a tight staleness cap. A Buffer cannot be ALTERed in place.
DROP TABLE IF EXISTS analytics.events_buffer;

CREATE TABLE analytics.events_buffer AS analytics.events_raw
ENGINE = Buffer(
    'analytics', 'events_raw',
    16,
    5,             -- min_time (s)
    15,            -- max_time (s): worst-case loss window shrinks to 15s
    100000, 1000000,
    10485760, 268435456
);

Lowering max_time narrows the exposure window but flushes smaller blocks during quiet periods, so it trades durability against part size. Keep it as tight as your part-size budget allows and commit offsets no earlier than max_time after the corresponding insert.

Step 4 — Add a circuit breaker on flush failure

The buffer blocks inserts once max_bytes is reached and retries failed flushes silently — rows pile up in RAM while writers keep succeeding until the ceiling hits. A circuit breaker watches resident memory and stops writers before that point, holding uncommitted offsets upstream instead of losing them.

python
def buffer_at_risk_bytes(client) -> int:
    row = client.query(
        "SELECT value FROM system.asynchronous_metrics "
        "WHERE metric = 'MemoryTracking'"
    ).result_rows
    return int(row[0][0]) if row else 0

MAX_BYTES_CEILING = 4 * 1024**3          # matches num_layers * max_bytes
TRIP_AT = int(0.7 * MAX_BYTES_CEILING)   # break at 70% of the ceiling

def guarded_insert(client, rows, columns):
    if buffer_at_risk_bytes(client) > TRIP_AT:
        # Do NOT insert or commit — let the source retain the offset and back off.
        raise RuntimeError("buffer near ceiling; pausing ingest to avoid loss")
    client.insert("events_buffer", rows, column_names=columns)

Because the writer refuses to insert (and therefore does not commit the offset) when the buffer is stressed, a subsequent crash replays cleanly from the last committed position. This is the same backpressure discipline the queue maxsize provides in the async writer pattern, applied to durability rather than throughput.

Step 5 — Replay the uncommitted tail on restart

On startup, the writer resumes from the last committed offset, which is at or behind the last durable row. At-least-once means some already-persisted rows get re-inserted, so the destination must absorb duplicates. Point replay at a ReplacingMergeTree keyed on a source id, or enable insert deduplication so identical blocks are dropped.

sql
-- Confirm replay did not double-count by comparing distinct source ids to total rows.
SELECT
    count()                       AS total_rows,
    uniqExact(source_event_id)    AS distinct_events,
    count() - uniqExact(source_event_id) AS duplicates_pending_merge
FROM analytics.events_raw
WHERE event_ts > now() - INTERVAL 1 HOUR;

duplicates_pending_merge above zero right after a restart is expected under at-least-once — a ReplacingMergeTree collapses them on the next merge. A persistently climbing figure means dedup is not configured correctly.

Verification

Simulate the failure. Insert a known batch, do not flush, and confirm the destination has not yet received it — those are the rows a crash would lose.

sql
-- Rows visible through the buffer but not yet durable on disk.
SELECT
    (SELECT count() FROM analytics.events_buffer) AS visible,
    (SELECT count() FROM analytics.events_raw)    AS durable;

Then drain and re-check: after OPTIMIZE TABLE analytics.events_buffer, durable should equal visible, proving the flush persisted everything. Track resident memory continuously with the counters in tracking memory pressure with system.metrics and alert well before max_bytes so the circuit breaker trips on warning, not on failure.

Gotchas & edge cases

  • A graceful shutdown flushes; a SIGKILL does not. SYSTEM SHUTDOWN and a clean service stop drain buffers before exiting, so a planned restart loses nothing. An OOM kill, kill -9, power loss, or a crashed container skips the drain and discards every resident row. Never assume restarts are graceful — size the exposure window for the crash case.
  • DETACH TABLE and DROP TABLE discard resident rows too. Detaching or dropping a buffer does not flush it first; the in-memory window is gone the moment the table leaves the server’s metadata. Always OPTIMIZE TABLE the buffer immediately before any detach, drop, or schema change that recreates it.
  • Silent flush retries mask a stuck destination. If the destination hits parts_to_throw_insert or runs out of disk, the buffer’s flush fails and retries in the background with no client-visible error — rows accumulate until max_bytes blocks inserts. Watch destination insert errors in system.query_log (type = 'ExceptionWhileProcessing') alongside buffer memory, so a failing flush surfaces before the ceiling does.
  • Committing offsets before flush breaks the whole guarantee. The most common data-loss bug is a consumer that auto-commits on a timer independent of flush state. Disable auto-commit and commit manually only after the drain in Step 2, or the exposure window silently becomes unrecoverable — the source has moved on from rows the buffer never persisted.

Up: Async Processing & Buffer Tables