Ops analytics from an event stream: Kafka into ClickHouse without losing or double counting alerts

Everything in this post is made up for illustration: the site, the cameras, the event ids, the timestamps, the counts. The problem is one we deal with building dashboards for Vision AI. On Aura every detection arrives with video evidence, gets routed to the team that acts on it, and rolls up into dashboards across sites. The routing side wants a transactional database: an alert gets raised, acknowledged, attached to a clip, closed. The dashboard side wants "PPE violations per dock per hour for the last 90 days", and asking the operational database for that during a busy shift is how alert pages get slow.

The pattern we'd recommend is a separate analytics path. Apps publish every alert and stat event to Kafka, a small consumer batches them into ClickHouse, and dashboards only ever read ClickHouse. The consumer is maybe 200 lines of Go. Almost all the interesting bugs live in about ten of them.

Here's the setup we'll follow.

site: plant-a (illustrative) cameras: gate-3 runs an intrusion app, dock-1 runs a PPE app topic: vision-events, 2 partitions, keyed by camera (gate-3 -> partition 0, dock-1 -> partition 1) consumer: group ops-analytics, buffers events, flushes to ClickHouse dashboard: alerts per camera per hour, open vs acknowledged 

The event and the table

One alert, as the intrusion app publishes it:

{"event_id": "evt-a1", "site": "plant-a", "camera": "gate-3", "app": "intrusion", "kind": "alert", "ts": "2026-03-14T02:11:07.412Z", "version": 1, "status": "raised"} 

Two fields matter more than the rest. event_id is minted by the app that saw the person at the gate, at the edge, before anything is published. version goes up when the same alert changes state. When an operator acknowledges evt-a1 a minute later, the app publishes the same event_id and ts with version: 2, status: "acknowledged". In our night shift that's six messages for five distinct alerts.

The table, as a sketch:

CREATE TABLE vision_events ( event_id String, site LowCardinality(String), camera LowCardinality(String), app LowCardinality(String), kind LowCardinality(String), ts DateTime64(3, 'UTC'), version UInt32, status LowCardinality(String) ) ENGINE = ReplacingMergeTree(version) PARTITION BY toYYYYMM(ts) ORDER BY (site, app, camera, ts, event_id); 

ReplacingMergeTree removes rows that share the same ORDER BY value, keeping the highest version (or the last inserted row if versions tie). The catch is in the docs, in plain words: deduplication only happens during background merges, at a time you can't plan for, and the engine "does not guarantee the absence of duplicates". If you need the right answer at query time you read with FINAL, which finishes the deduplication while the query runs.

So the sorting key is your dedup key, and three rules fall out of that. Every column in ORDER BY has to be identical across copies of the same event, which is why ts is the event time from the camera and not the time the consumer saw it. The partition key has to be identical too, since merges happen within a partition. And event_id has to come from the source. A consumer that mints a fresh UUID per row creates a new key on every replay, and no merge will ever collapse those.

The consumer loop

The shape is boring on purpose. Poll, append to a buffer, flush when the buffer reaches a size or when its oldest event reaches an age, then move offsets.

flush when: buffer >= MaxBatch (500) or oldest buffered event >= MaxWait (2s) then: INSERT batch -> commit offsets (last offset + 1 per partition) 

Batching is not optional with ClickHouse. Each synchronous insert creates a part on disk, and the docs recommend batches of at least 1,000 rows, ideally 10,000 to 100,000, and roughly one insert query per second. 500 is below that minimum, which is fine for a single quiet site where the 2-second timer does most of the flushing and inserts stay under one per second. A busier deployment should push MaxBatch up. If you can't batch on the client, async inserts move the buffering to the server. If you go that way keep wait_for_async_insert=1, so a failed flush still comes back to the consumer as an error.

The order of "insert" and "commit" in that second line is what the rest of this post is about.

Example 1: commit first, insert times out, pod dies

An illustrative timeline:

02:14:00.000 consumer has 6 events buffered (3 per partition), flushes 02:14:00.001 offsets committed: p0=3 p1=3 02:14:10.002 INSERT times out (ClickHouse overloaded) 02:14:11 pod OOMKilled before the retry 02:14:40 new pod joins the group, resumes at p0=3 p1=3, nothing to read 

The toy reproduces it (go run ./cmd/demo, scenario 1):

commit p0=3 p1=3 insert 6 rows failed: insert timeout (will retry) -- process restarts, resumes from committed offsets -- rows: count()=0 FINAL=0 inserts=0 GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG ops-analytics vision-events 0 3 3 0 ops-analytics vision-events 1 3 3 0 

Six alerts are gone from the dashboard and lag says zero. Kafka is right, the group did consume them. Every monitor that only watches lag stays green.

Nobody writes "commit, then insert" on purpose. It shows up through defaults. With librdkafka based clients (confluent-kafka-go included), enable.auto.commit and enable.auto.offset.store are both true by default, which stores each message's offset as it's handed to your code and commits it from a background thread on a timer. That commit has no idea your batch is still sitting in memory. The librdkafka configuration docs recommend setting enable.auto.offset.store=false for this kind of app and storing offsets yourself after processing.

Before and after, as a sketch with confluent-kafka-go and clickhouse-go (error handling trimmed):

// before (illustrative): defaults, offsets move as messages are read cfg := &kafka.ConfigMap{"bootstrap.servers": brokers, "group.id": "ops-analytics"} // after: offsets only move once ClickHouse said OK cfg := &kafka.ConfigMap{ "bootstrap.servers": brokers, "group.id": "ops-analytics", "enable.auto.offset.store": false, } func (s *sink) flush(ctx context.Context) error { batch, _ := s.ch.PrepareBatch(ctx, "INSERT INTO vision_events") for _, e := range s.buf { batch.AppendStruct(&e) } if err := batch.Send(); err != nil { return err // keep buf and offsets, retry next tick } // enable.auto.commit stays true (the default): StoreOffsets only stores; // a background thread still commits those stored offsets. StoreOffsets // itself does not commit. _, err := s.kc.StoreOffsets(s.nextOffsets()) // last offset + 1, per partition s.buf = s.buf[:0] return err } 

Setting enable.auto.commit=false and calling Commit right after Send works just as well. The Java consumer docs walk through the same batch-then-commit loop and call the result at-least-once, meaning a crash in the wrong place gets you duplicates.

Example 2: insert first, pod dies before the commit

Same night, offsets now move after the insert.

02:14:00.000 flush 6 events, INSERT ok 02:14:00.040 pod killed (node drained) before the commit 02:14:30 new pod resumes at p0=0 p1=0, reads the same 6 messages, INSERT ok, commit 

Toy, scenario 2:

insert 6 rows ok -- process restarts, resumes from committed offsets -- insert 6 rows ok commit p0=3 p1=3 rows: count()=12 FINAL=5 inserts=2 ... FINAL evt-a1: version=2 status=acknowledged 

Nothing lost, everything twice. A plain count() says 12, and with FINAL it's 5, because the two copies of each event share a sorting key and the acknowledged evt-a1 beats its raised version. That's the better failure to have, since duplicates can be cleaned up and missing alerts can't.

ClickHouse also has insert deduplication on retries: each inserted block gets a hash, and a retry of an identical block is dropped. It's enabled by default for replicated tables (a plain MergeTree needs non_replicated_deduplication_window set). It helps when the same process retries the same batch after a timeout. It does much less after a restart, because the new pod cuts batches at different boundaries (the 2-second timer fires at a different moment) and the hashes differ. You can pass an insert_deduplication_token built from partition and offset range if you cut batches per partition deterministically, but we treat the idempotent table as the backstop and the token as an optimization.

Dashboards then read with FINAL, or aggregate in a way that ignores copies:

-- illustrative: open vs acknowledged alerts per camera per hour, last 24h SELECT camera, toStartOfHour(ts) AS hour, countIf(status = 'raised') AS open, countIf(status = 'acknowledged') AS acked FROM vision_events FINAL WHERE site = 'plant-a' AND kind = 'alert' AND ts >= now() - INTERVAL 1 DAY GROUP BY camera, hour ORDER BY hour, camera; 

FINAL costs something on big scans. Filtering on the leading ORDER BY columns (site, and app when you can) keeps it cheaper, and plain counts of distinct events can use uniqExact(event_id) without FINAL at all. The query above filters on kind too, which is useful but isn't in the key, so it doesn't help the scan the same way.

A test in the toy pins down what FINAL can't fix. An exact resend collapses. A replay with a re-minted id, or with ts restamped at ingest, stays as a separate row forever.

Example 3: SIGTERM during a rollout

A deploy rolls the consumer. Kubernetes sends SIGTERM and gives the pod 30 seconds by default before SIGKILL. At that moment the buffer holds our six events, waiting for the 2-second timer.

// before (illustrative): stop on signal, close, done <-sigterm consumer.Close() 

With the default librdkafka settings from example 1, Close() commits the offsets it has stored on the way out, and those already cover the buffered messages. Toy scenario 3a:

shutdown: closing with 6 rows buffered rows: count()=0 FINAL=0 inserts=0 GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG ops-analytics vision-events 0 3 3 0 ops-analytics vision-events 1 3 3 0 

The fix is an ordered shutdown: stop polling, flush, move offsets, then close.

// after: final flush, then commit, then close ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, os.Interrupt) defer stop() for ctx.Err() == nil { s.step(ctx) // poll + flush on size/time } flushCtx, cancel := context.WithTimeout(context.Background(), 20*time.Second) defer cancel() if err := s.flush(flushCtx); err != nil { log.Printf("final flush failed, offsets left for replay: %v", err) } s.kc.Close() 

Toy scenario 3b:

shutdown: final flush of 6 rows insert 6 rows ok commit p0=3 p1=3 rows: count()=6 FINAL=5 inserts=1 

Two details. The final flush gets its own timeout, shorter than the grace period, so a hung ClickHouse can't eat the whole 30 seconds. And if that flush fails, offsets stay put, the next owner replays the batch, and example 2 handles the copies. If your offsets only ever move after an insert, a skipped final flush costs a replay. With auto-stored offsets it costs data. The same flush belongs in the rebalance callback, when partitions are revoked from a running consumer, for the same reason.

Example 4: ClickHouse is down and the cameras keep sending

Say ClickHouse is unreachable for a few minutes. The cameras don't care, and alerts keep coming. A consumer that keeps polling and appending will grow its buffer until the pod gets OOMKilled, which is example 1 again if anything committed early.

So the buffer gets a ceiling (MaxBuffered). Above it the consumer stops taking new records and keeps retrying the flush. With a real Kafka client, pause the assigned partitions (Pause) and keep calling Poll. If you stop polling for longer than max.poll.interval.ms, the client leaves the group and a rebalance follows.

Toy scenario 4, with a buffer ceiling of 4 and the store failing every insert while two cameras trickle in stat events:

insert 4 rows failed: insert timeout (will retry) paused: 4 buffered >= 4 insert 4 rows failed: insert timeout (will retry) paused: 4 buffered >= 4 insert 4 rows failed: insert timeout (will retry) GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG ops-analytics vision-events 0 0 15 15 ops-analytics vision-events 1 0 15 15 -- store back -- paused: 4 buffered >= 4 insert 4 rows ok commit p0=2 p1=2 ... commit p0=15 p1=15 rows: count()=30 FINAL=29 inserts=8 

count()=30 is the six night-shift messages plus 24 stat events (three Trickle ticks of four per partition). FINAL=29 because evt-a1's raised and acknowledged versions collapse again, same as in example 2.

The backlog now sits in Kafka, which is built to hold it (as long as the outage is shorter than the topic's retention), and drains once inserts succeed. Real kafka-consumer-groups --describe output also shows CONSUMER-ID, HOST and CLIENT-ID. A slow catch-up might look like this (illustrative):

$ kafka-consumer-groups --bootstrap-server kafka:9092 --describe --group ops-analytics GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID ops-analytics vision-events 0 418200 431950 13750 ops-analytics-1-7c2e /10.0.4.17 ops-analytics-1 ops-analytics vision-events 1 402115 402130 15 ops-analytics-1-7c2e /10.0.4.17 ops-analytics-1 

Read lag as a trend. A steady small number is normal, a number that grows between two checks a minute apart means inserts are failing or too slow. One hot partition, like partition 0 above, usually points at one camera or one app flooding events, since the topic is keyed by camera.

Example 1 showed lag can't be the only signal. We'd pair it with two numbers from the consumer itself: rows inserted per minute and time since the last successful insert. Lag at zero with no inserts for ten minutes during a shift is the silent loss case, and only the second number (time since the last insert) catches it. A quiet night also has zero inserts, so to avoid that false alarm compare against log-end-offset growth: events arriving on the topic while nothing gets inserted.

Tradeoffs

You run one more system and one more consumer. For a single small site, a read replica of the operational database plus a few rollup tables may be enough, and we'd start there.

Dashboards are seconds behind. Fine for trends and compliance rates. Alert routing should stay on the operational path, where an operator needs the event now.

At-least-once plus a ReplacingMergeTree means duplicates exist on disk until a merge, so every query has to be written with that in mind (FINAL, uniqExact, or argMax on version). It's easy for someone to add a dashboard panel with a plain count() six months later, so put a view in front of the table or at least a note in its comment.

Schema changes need care in both directions. Add columns with defaults, and let the consumer ignore fields it doesn't know yet, so you can roll apps and the consumer in either order.

Run the toy

The toy is plain Go with no dependencies outside the standard library. sink/broker.go is a fake broker with partitions, committed offsets, restart and a describe table, sink/store.go is a fake ReplacingMergeTree with Count() and Final(), and sink/consumer.go is the batching consumer with both commit orders, the buffer ceiling and the two shutdown paths. Tests cover all four examples, flush on size and time, a failed final flush leaving offsets for replay, a bad JSON record, and the re-minted id case.

cd toy go test -v ./... go run ./cmd/demo 

Every block introduced above as a toy scenario comes from go run ./cmd/demo. Flip Mode or FinalFlush in cmd/demo/main.go and watch the counts move.