Exactly-once doesn’t come for free: Kafka, Flink and ClickHouse

"Exactly-once" is an easy phrase to put in a design doc. Flink has exactly-once checkpoints, Kafka has idempotent producers, ClickHouse has tables that deduplicate themselves. Put the three together and the data feels safe.

This post walks through an event analytics pipeline I started and never finished: webhooks from GitHub and Jira hit a gateway, go through Kafka, get aggregated by Flink and land in ClickHouse. Rereading the code for this post, I found that every stage had a place where data could be lost or counted twice, even though the design doc said otherwise. Each one is a fairly common mistake, so here they are with the fixes.

Three of the stages could lose or double-count data even though each tool advertises exactly-once.

One principle runs through all of it: end-to-end exactly-once is almost always at-least-once plus an idempotent destination. The question at each stage is: if this step runs again, does the result stay the same?

Stage 1: the gateway claims the key before the work is done

Webhook providers retry when they don't get a success response, so the gateway needs to deduplicate. The first version:

if (!idempotencyStore.claim(event.idempotencyKey())) {   // Redis SET NX
    return duplicate();
}
publisher.publish(event);   // async send to Kafka, not awaited
return accepted();          // 2xx right away

The problem is the order. The key is claimed before the event is actually in Kafka, and the gateway answers 2xx without waiting for Kafka to acknowledge. If Kafka fails at that moment:

1. The provider got a 2xx, so it doesn't retry.
2. If it does retry, the retry is treated as a duplicate, because the key is already in Redis.

The provider's retry, the cheapest recovery there is, gets wasted. What's left is the dead letter queue (DLQ), which usually lives on the same Kafka cluster that just failed.

Fixes, depending on how strict you need to be:

• Wait for Kafka's acknowledgement before answering 2xx. This is the most important change. On failure, answer 5xx so the provider retries.
• Only mark the key "done" after a successful send, or use two states: pending when starting (with a short TTL), done once Kafka acknowledges, and delete the key on failure.
• If an exact deduplication layer exists downstream (here, Flink, see below), the gateway's layer only needs to be an optimization, not a guarantee. Two concurrent deliveries slipping through is fine, because Flink drops the second one.

The general lesson for any idempotency key: a key may only say "done" once the side effect is done. Claiming it before the side effect turns a transient error into permanent data loss.

Stage 2: Flink deduplicates with keyed state

In Flink, events are keyed by orgId:source:eventId, and a ValueState remembers the keys it has seen:

Long firstSeen = firstSeenState.value();
if (firstSeen == null) {
    firstSeenState.update(event.eventTime().toEpochMilli());
    out.collect(event);
}
// already seen: drop, increment dedup_dropped_total

This stage is right. The state is part of the checkpoint, so after a restart from a checkpoint, the Kafka offsets and the set of seen keys go back to the same point in time. The state has a TTL so it doesn't grow forever, and the TTL is extended only on write, not on read (OnCreateAndWrite): a duplicate arriving 23 hours later is still caught, and checking a key doesn't keep it alive longer.

The thing to remember is that the dedup window must cover the longest gap between two deliveries of the same event (say, how long the provider keeps retrying). It has nothing to do with how long Kafka retains data. A few easily misread details of state TTL are in time in Apache Flink.

Stage 3: a buffering sink that isn't tied to checkpoints

Writing to ClickHouse row by row is far too slow, so the sink batches 1,000 rows. The first version extended RichSinkFunction, kept a List as its buffer, and had this comment:

// Retries exhausted: log and drop the batch. The checkpoint won't advance,
// so the batch will be retried when the job restarts.

The comment is wrong, in a way that's very easy to get wrong:

• A Flink checkpoint snapshots operator state. A plain List in a sink isn't state, and the sink does nothing when a checkpoint happens. The barrier passes through, the checkpoint completes and Kafka offsets move on, while those rows are still only in memory.
• If the job dies now, restarting from the checkpoint resumes after those rows. They're gone.
• A batch "dropped" after its retries is the same: nothing stops the checkpoint from advancing, so that batch is never written.

One more detail: the comment says batches are flushed every 5 seconds "by a background thread", but the code only checks the time when a new record arrives. On a quiet stream, data sits in the buffer indefinitely.

The fix:

public class ClickHouseSink extends RichSinkFunction<Row> implements CheckpointedFunction {
    @Override
    public void snapshotState(FunctionSnapshotContext ctx) throws Exception {
        flush();   // drain the buffer before the checkpoint may complete
    }

    private void flush() throws Exception {
        // bounded retries; when exhausted, THROW instead of dropping,
        // so the checkpoint fails and the job restarts from the previous one
    }
}

• Flush in snapshotState (from CheckpointedFunction): a checkpoint only completes once every row before the barrier is in ClickHouse. That's what at-least-once requires.
• Throw instead of dropping: the job restarts and rewrites from the previous checkpoint. You may get duplicates, but nothing is lost, and the next stage can handle duplicates.
• Time-based flushing needs a real timer, plus a lock between the timer thread and the processing thread.
• Best of all, don't write your own: Flink's official JDBC connector (JdbcSink with JdbcExecutionOptions) and the AsyncSinkBase foundation for new sinks already handle batching, timers, retries and flushing on checkpoint.

Stage 4: ReplacingMergeTree deduplicates, eventually

An at-least-once sink means ClickHouse will receive duplicates. The tables use ReplacingMergeTree(checkpoint_version), with checkpoint_version set to the window's end time, so recomputing the same window produces exactly the same row with the same version.

What you need to know: ReplacingMergeTree deduplicates when parts merge in the background, not on insert. Until then, SELECT sees both copies. Reads must use FINAL, or group by key and take the highest version. The project gets this right: the queries on the deployment and incident tables, the two ReplacingMergeTree tables, all use FINAL.

Stage 5: a materialized view double-counts, and merges can't fix it

This is the hardest one to see. The normalized events table is a ReplacingMergeTree, and a materialized view feeds hourly rollups from it:

CREATE MATERIALIZED VIEW mv_employee_metrics_hourly TO employee_metrics_hourly AS
SELECT org_id, employee_id, toStartOfHour(event_time) AS window_end,
       sumState(toUInt32(event_type = 'pr.merged')) AS pr_merged,
       ...
FROM normalized_events

A ClickHouse materialized view is a trigger on each inserted block, not a query over merged data. Insert an event twice and the view sees it twice, so the rollup counts it twice. When ReplacingMergeTree later collapses the two copies in the source table, the rollup never finds out. The "idempotent" source table doesn't carry its idempotency into the rollup.

The view fires on every inserted block, so a merge in the source table never reaches the rollup.

A few ways out, depending on what you need:

• Aggregate in Flink, have ClickHouse store results. Flink already has windows; write each window's result straight into a ReplacingMergeTree keyed by window. Writing it again produces the same row.
• Deduplicate at insert time: ClickHouse can skip identical inserted blocks (by default for replicated tables; plain tables need non_replicated_deduplication_window above 0), and extends that to dependent views through deduplicate_blocks_in_dependent_materialized_views (on by default only since 26.2, so on the 24.3 this pipeline ran it has to be turned on). A retry must resend the exact same block, or carry the same insert_deduplication_token, for example built from the checkpoint id and subtask.
• Recompute on a schedule from the source with FINAL (a refreshable materialized view), accepting the delay.

Bonus: zero isn't always neutral

The same view fills columns that don't have data yet with placeholders:

sumState(toUInt32(0))   AS review_comments,       -- harmless
avgState(toFloat64(0))  AS pr_review_latency_s,   -- not harmless

Adding 0 to a sum leaves it unchanged. Adding 0 to an average gives every event a sample of zero, dragging the average down. The real values Flink writes into the same table get diluted by thousands of zeros. It's also why AggregatingMergeTree stores an average's state (sum and count) rather than the average itself: you can't add averages together.

Every aggregate function has its own neutral element: for sum it's 0, for avg it's "no samples". To reserve a column for an average, produce an empty state, for example avgStateIf(toFloat64(0), 0), or leave the column out of the view until real data exists.

Summary

Stage	Risk	Fix
Gateway	Key claimed before the send completes	Wait for Kafka's ack before 2xx; mark done only when done
Flink	(correct)	Dedup with keyed state with TTL, inside the checkpoint
Sink	Buffer not tied to checkpoints	Flush in snapshotState, throw instead of dropping, or use a built-in sink
ClickHouse	Dedup happens at merge time	Read with FINAL
Materialized view	Counts duplicates before any merge	Aggregate in Flink, or deduplicate inserted blocks

None of these stages used the "wrong" technology. The bugs live in the joints, where each component keeps its own promise but nobody keeps the promise for the whole chain. What I find most useful when designing is to walk every arrow in the diagram and ask: what if this step runs again, or dies halfway?

A similar pipeline may still have gaps I haven't spotted. If you've run into one, I'd love to hear about it. My email is on the home page.