Ordering and Checkpoints: Out-of-Order Arrival and the Bounded Apply

What the log actually promises

Kafka guarantees order within a partition. That is the whole promise. There is no global ordering across a topic, and there never will be, because partitions are the unit of parallelism and a total order across them would mean a single consumer.

So ordering is a consequence of the partition key you choose. Key the change events by the row's primary key and every change for that row hashes to the same partition and arrives in log order, forever. Key them by anything else, and the events for one row scatter across partitions whose consumers run at different speeds. Then a later change can be applied before an earlier one, and the row ends up holding a stale value with no error anywhere.

The changelog in this lesson came off a connector keyed by warehouse_id. Orders move between warehouses, so one order's events landed on different partitions, and three keys arrived inverted.

Table

Arrival order is an accident of partitioning. The lsn column is the truth, and the checkpoint is a line drawn on it, not on arrival_seq.

arrival_seq	order_id	lsn	after_status	reading
3	1042	1207	shipped	arrived first
5	1042	1201	packed	OUT OF ORDER: lower lsn arrived later
12	1043	1210	shipped	checkpoint line: apply stops here, and it arrived 12th
19	1042	1217	delivered	beyond the checkpoint, next run's work

Detecting it before it bites

Walk each key in arrival order and compare each event's lsn to the previous one's. LAG(lsn) OVER (PARTITION BY order_id ORDER BY arrival_seq) gives you that previous value; wherever the current lsn is SMALLER, the stream handed you the past after it handed you the future. That query is worth keeping as a monitor, because it is the only cheap way to notice a bad partition key before someone notices a wrong dashboard.

The fix at the sink is never "hope". Sort by lsn at apply time. Arrival order is an accident; the log position is a fact.

The apply, in three moves

You have built these three before, in sql-l5-cdc-changelog-apply, on a changelog ordered by a plain version number. They are recalled here as a component, not retaught: what is new in this lesson is that the order comes from a log position you do not control, and that the apply is bounded by a checkpoint it has to record and resume from.

1. Reduce. Keep one event per key, the one with the highest lsn. Same ROW_NUMBER reduce as the previous lesson.
2. Delete. Keys whose winning event is op = 'd' are gone. Remove them from the target.
3. Upsert. Every other key writes its after image, inserting when the key is new and updating when it already exists.

Interviewers call this whole thing a MERGE, and on Snowflake, BigQuery, and Athena-on-Iceberg you would literally write MERGE INTO target USING source ON ... WHEN MATCHED THEN UPDATE ... WHEN NOT MATCHED THEN INSERT .... SQLite has no MERGE keyword, so you write the same semantics as INSERT ... ON CONFLICT(order_id) DO UPDATE SET status = excluded.status plus a keyed DELETE for the tombstone branch. That is not a workaround to apologize for in an interview. A merge is an upsert plus a delete branch, and being able to say that sentence is the understanding the question is testing. You have already graded the upsert mechanics in de-l8-change-batch-merge; here they are a component, not the lesson.

One parser detail that costs people ten minutes: when the rows you insert come from a SELECT rather than a VALUES list, SQLite cannot tell the ON of ON CONFLICT from the ON of a join, and it reports near "DO": syntax error. The fix is in SQLite's own documentation: give the SELECT a WHERE clause, even a trivial WHERE true, so the parser has something between the source and the conflict clause. A reduce that already filters WHERE rn = 1 satisfies this without you noticing.

The checkpoint

A consumer that runs forever does not exist. Yours will be restarted, redeployed, and rerun, so it needs to remember how far it got. That memory is a watermark: after applying every event with lsn <= C, record C in a small control table, and next run start from there.

Two properties matter and both are graded here.

• The bound is on the log position, not on time and not on arrival order. lsn <= 1210 describes a slice that means the same thing on every rerun, on every replay, and on any consumer. The other two bounds cannot even name this slice: lsn 1210 is the twelfth arrival, not the tenth, and it shares a commit millisecond with lsn 1214, so a timestamp bound either drags 1214 in or drops 1210.
• The apply must be idempotent, because at-least-once delivery guarantees you will be handed the same slice twice. Running the script a second time has to leave orders and apply_watermark exactly as the first run left them. The grader checks this literally, by running your script twice.

That second property is why the watermark is never written with a bare insert: a bare insert stacks a second row on the rerun and the checkpoint stops being a single number. Which write is right depends on whether the row exists yet. An UPDATE is the natural rerun-safe write, but it only rewrites a row that is already there, so it does nothing at all on a control table that starts empty, and the practice's apply_watermark starts empty. A DELETE followed by an INSERT is the shape that is correct on the first run and on every run after it.

Common mistake: applying in arrival order because the events "usually" arrive in order. They usually do. The one time they do not, order 1042 ends at packed when the source says shipped, nothing errors, and the discrepancy is found weeks later by a human who noticed a stuck order.

Interview nuance: "How do you guarantee ordering in Kafka" has a two-word core answer, partition key, and the correct qualifier that order holds within a partition only. The follow-up is always "and what if it is already shuffled by the time you get it", and the answer is that you order on the log position at the sink, because the sink is the last place that can still be made correct.

On a real platform this differs. Here the watermark is a one-row SQLite table you write yourself. A real consumer commits Kafka offsets back to the broker, a Flink job checkpoints its state to durable storage, and a warehouse-side merge job keeps its high-water mark in a control table that looks very much like this one. Delta Lake, Iceberg, and Hudi all give you an atomic MERGE so the delete branch and the upsert branch commit together; in SQLite they are two statements you would wrap in a transaction. The reduce-by-log-position step is identical everywhere, including in Flink, where it is the job's keyed state rather than a SQL window.