Consumer offsets and commit ordering

Why is the order of the offset commit the delivery semantic, and how does storing the offset in the sink remove the choice?

This is the smallest possible instance of the whole article, so it is the best teaching case. A consumer reads from the log, does something, and records how far it has read. There are two writes: the offset (to the broker) and the effect (to the sink). They live in two stores with no shared transaction.

There are two orderings, and the ordering is the delivery semantic:

ORDER A  (offset first)           ORDER B  (effect first)
 1. commit offset 5215             1. apply effect of record 5215
 2. apply effect of record 5215    2. commit offset 5215
    crash between 1 and 2 ->          crash between 1 and 2 ->
    record 5215 is LOST               record 5215 is REPROCESSED
    = at-most-once                    = at-least-once

No third ordering exists. But there is an escape that is not an ordering, and it is the correct answer: store the offset in the sink. If the position is written inside the same transaction as the effect, there are no longer two writes. The problem does not need solving, because it does not exist.

BEGIN;
INSERT INTO patient_records (result_id, patient_id, payload) VALUES (...);
UPDATE consumer_offsets
   SET next_offset = 5216
 WHERE topic = 'lab-results' AND partition = 3 AND next_offset = 5215;
COMMIT;   -- both happen, or neither
-- on startup: read next_offset from this table and seek the log to it

Writing to PostgreSQL and keeping the offset in a PostgreSQL row is the cheapest exactly-once available anywhere. It needs no engine feature. It is chronically underused, because the transport ships its own offset store and people assume they should use it. The reason it works: where a system lacks a single atomic primitive across two stores, some external component appears to supply one. Here the sink already is one.

A diagnostic follows, and it classifies most pipelines in ten seconds:

Four details that hand-written consumers often get wrong:

Auto-commit is a silent semantics decision

Most client libraries have, by default, a periodic background commit. It commits whatever offset has been fetched, on a timer, without knowing whether processing finished.

This gives you at-most-once that looks like at-least-once. Records that were fetched but not yet processed get their offsets committed by the timer. If the process crashes, they are gone: no error, no metric, no evidence. It is the most common silent-data-loss setting in practice, and it is a default.

Check whether auto-commit is on. If it is, know that you chose at-most-once.

Commit frequency sets duplicate volume

If you commit the offset every N records, a failure replays up to N records. Duplicates are not a yes/no property. They are a quantity you choose: commit interval × rate. Example: commit every 5 s at 2,000 records/s means up to 10,000 duplicates per crash. More frequent commits cost more writes to the broker. This is the same batching arithmetic as before (pay a fixed cost less often), with the duplicate count as the thing being traded.

Out-of-order completion makes naive offset commit unsound

If a consumer processes records in parallel or asynchronously, record 60 may finish before record 50. If you commit "the highest completed offset" (60), a crash silently loses record 50. The correct rule is to commit the low-water mark: the offset below which everything is complete.

Example. Records 50 to 60 are in flight. Records 51 to 60 are done, 50 is still running. The low-water mark is 49 (everything up to 49 is complete), so you commit 50 as "next to read". If you crash now, you re-read 50 to 60. Records 51 to 60 become duplicates, but nothing is lost.

This requires tracking a set of in-flight offsets. It also means your duplicate window is bounded by the slowest in-flight record, not by the batch.

Offset commits must be fenced

After a rebalance (the group moves a partition to a different consumer), the partition belongs to someone else. A commit from the previous owner is a stale write. This is the fencing token idea again: the consumer generation is the epoch, and a commit from an old generation must be rejected by the resource that stores the offset. A consumer that catches and logs the rebalance exception and carries on is corrupting positions.