Transactions and read_committed

After this lesson your consumer can tell committed data from aborted and unfinished data in a log where transactions from several producers are interleaved, and it stops at the last offset that is safe to read.

What a transaction is for

Idempotence gives one producer exactly-once writes to one partition. A stream processor needs more: read from topic A, write results to topics B and C, and record the input position, as one all-or-nothing step. Kafka transactions provide that. A producer with a stable transactional.id does:

beginTransaction
  send to B, send to C, sendOffsetsToTransaction(input offsets)
commitTransaction        (or abortTransaction)

Behind it, a transaction coordinator (a broker, with its state in the internal __transaction_state topic) tracks which partitions the transaction touched. The producer id and epoch are tied to the transactional.id, so a restarted instance gets a higher epoch and the old one is fenced.

The important design choice: write first, decide later

The records are appended to the partitions immediately, mixed with everybody else's, flagged as transactional. Nothing is held back on the broker. When the transaction ends, the coordinator writes a marker into every partition it touched. So the log looks like this:

offset:  0        1        2        3         4      5        6
         P7 data  P8 data  plain    P8 COMMIT P7 data plain    ...

The consumer, not the broker, decides what to show.

How the log says what is what

All of it is in the batch header's attributes field (an int16 at byte 21):

A marker is a normal batch with one record. Its key is two int16s, version and type (0 = ABORT, 1 = COMMIT); its value is a version and the coordinator epoch. The batch's producerId says whose transaction ends. Markers consume an offset, which is why a consumer sees gaps in offsets, and they are never given to the application.

Isolation levels and the LSO

• read_uncommitted (the default) delivers every data record up to the high watermark, aborted ones included.
• read_committed must not deliver records of an aborted transaction, nor of one that has not finished. For that it reads only up to the last stable offset (LSO): the first offset of the earliest transaction that is still open, or the high watermark if none is. Below the LSO every transaction has a marker, so every record's fate is known.

Consequence: one open transaction blocks everything behind it, including plain records and already-committed transactions of other producers. That is why long transactions hurt consumers, and why transaction.timeout.ms exists: the coordinator aborts a transaction that does not finish in time.

Worked trace

Different data from your exercise. HW is 6, and the log is:

0  P7 txn  "x"
1  P8 txn  "y"
2  plain   "z"
3  P8 COMMIT
4  P7 txn  "w"
5  plain   "u"

P7's transaction (offsets 0 and 4) has no marker, so it is open and its first offset is 0. The LSO is 0: a read_committed consumer sees nothing, although y is committed and z and u were never transactional. A read_uncommitted consumer sees x y z w u.

Now append 6 P7 ABORT and set the HW to 7. There are no open transactions, the LSO is 7, and P7's records are known to be aborted:

read_committed:   1 y,  2 z,  5 u          (x and w are hidden; markers never shown; end=7)

Common mistakes

• Showing records because the batch is "transactional" and a COMMIT appears somewhere. The marker has to be the one of the same producer id, after the batch.
• Reading past the LSO because the later records look fine.
• Forgetting that a transaction that began before your fetch offset still decides the fate of records after it.
• Surfacing markers. They are control data, not records.
• One transaction per producer, forever. After a marker, the same producer id starts a fresh transaction with its next batch.

Your exercise

The starter splits the log into batches for you. Implement the record decoding and the delivery rules. The most likely mistake is to compute the LSO from the first transactional batch in the log instead of from the earliest transaction that is still open.