KafkaIQ

8.1 Transaction & Exactly-Once Internals (Broker Side)

Module Overview

Goal: Explain broker-side transaction and exactly-once semantics (EOS) internals.

What You'll Learn:

  • Transaction coordinator internals
  • Transaction log (__transaction_state)
  • Producer IDs and epochs
  • Transaction markers (commit/abort)
  • Exactly-once semantics (EOS) at broker level
  • Transactional reads and isolation levels

What's NOT Covered Here:

  • Producer API usage (covered in Producer modules)
  • Kafka Streams EOS (covered in Streams modules)
  • Consumer transaction handling (covered in Consumer modules)

Key Insight: Transactions in Kafka are broker-coordinated using a transaction coordinator, transaction log, and commit/abort markers. Understanding broker-side mechanics explains how Kafka achieves exactly-once semantics.

1. Why Transactions?

The Problem

WITHOUT TRANSACTIONS:

Scenario: Move money between accounts (read from account A, write to account B)

1. Producer reads from topic "accountA"
2. Producer processes data
3. Producer writes to topic "accountB"
4. Producer crashes before committing offset

Result:
├─ Data written to "accountB" (duplicate)
├─ Offset not committed
└─ Reprocessing creates duplicate write

Exactly-once goal:
└─ Write to accountB and commit offset atomically
└─ Either both succeed or both fail (no duplicates)

Kafka Transactions

WITH TRANSACTIONS:

1. Producer begins transaction
2. Producer writes to topic "accountB"
3. Producer sends offset commit to transaction (not consumer group)
4. Producer commits transaction (atomic)

Result:
├─ If commit succeeds: Write and offset both visible
├─ If commit fails: Write and offset both discarded
└─ Atomic "all-or-nothing" semantics

Guarantees:
└─ Exactly-once processing (no duplicates, no data loss)

2. Transaction Coordinator

Role and Responsibility

TRANSACTION COORDINATOR:

Purpose:
└─ Manage transaction state and lifecycle

One coordinator per transaction:
└─ Determined by hashing transactional.id

Responsibilities:
├─ Assign Producer IDs (PIDs)
├─ Track transaction state (ongoing, prepared, completed)
├─ Write transaction state to transaction log
├─ Send transaction markers (commit/abort) to partitions
└─ Handle coordinator failover

Coordinator is:
└─ A broker designated for a range of transactional IDs
└─ Not a separate process (part of broker)

Coordinator Selection

DETERMINE COORDINATOR:

transactional.id = "order-processor-1"

1. Hash transactional.id:
   └─ hash("order-processor-1") % num_transaction_partitions

2. Result: Partition X in __transaction_state topic

3. Leader of partition X = Transaction coordinator

4. Producer sends FindCoordinatorRequest:
   └─ "Who coordinates transactional.id='order-processor-1'?"

5. Broker responds:
   └─ "Coordinator is Broker 2"

6. Producer connects to Broker 2 for transaction operations

3. Producer ID (PID) and Epoch

PID Assignment

PRODUCER ID (PID):

Definition:
└─ Unique 64-bit ID assigned to producer by coordinator

Assignment:
1. Producer sends InitProducerIdRequest to coordinator
   └─ Includes transactional.id (if transactional)

2. Coordinator:
   ├─ Checks if transactional.id exists
   ├─ If new: Assign new PID
   ├─ If existing: Return existing PID (and increment epoch)
   └─ Write to transaction log

3. Coordinator responds:
   └─ PID: 12345
   └─ Epoch: 0

4. Producer uses PID in all subsequent produce requests

Purpose:
└─ Identify producer uniquely
└─ Enable idempotence and transactions

Producer Epoch

EPOCH FOR FENCING:

Epoch increments on:
├─ Producer restart (same transactional.id)
├─ Transaction coordinator failover
└─ Explicit fence (e.g., via admin API)

Example:
├─ Producer 1 (PID=12345, epoch=0) starts
├─ Producer 1 crashes
├─ Producer 2 (same transactional.id) starts
├─ Gets PID=12345, epoch=1
└─ Old producer (epoch=0) is fenced

Fencing:
└─ Brokers reject requests from lower epochs
└─ Prevents zombie producers from writing

4. Transaction Log (__transaction_state)

Log Structure

__transaction_state TOPIC:

Internal Kafka topic:
├─ 50 partitions (default, configurable)
├─ Replication factor: 3 (configurable)
├─ Compacted (only latest transaction state per key)
└─ Not visible to regular consumers

Key:
└─ transactional.id

Value:
└─ Transaction metadata:
    ├─ PID (Producer ID)
    ├─ Epoch
    ├─ Transaction state (Ongoing, PrepareCommit, CompleteCommit, etc.)
    ├─ Transaction timeout
    ├─ Partitions involved in transaction
    └─ Timestamp

Transaction States

TRANSACTION STATE MACHINE:

Empty:
└─ No active transaction

Ongoing:
└─ Transaction in progress (writes happening)

PrepareCommit / PrepareAbort:
└─ Producer requested commit/abort, coordinator preparing

CompleteCommit / CompleteAbort:
└─ Markers written, transaction complete

Dead:
└─ Transaction timed out or explicitly aborted

STATE TRANSITIONS:

Empty → Ongoing:
└─ Producer begins transaction (first write)

Ongoing → PrepareCommit:
└─ Producer calls commitTransaction()

PrepareCommit → CompleteCommit:
└─ Coordinator writes commit markers to all partitions

Ongoing → PrepareAbort:
└─ Producer calls abortTransaction() or timeout

PrepareAbort → CompleteAbort:
└─ Coordinator writes abort markers

CompleteCommit/CompleteAbort → Empty:
└─ Transaction finalized, ready for next transaction

5. Transaction Flow

Beginning a Transaction

START TRANSACTION:

1. Producer configured with:
   └─ transactional.id = "order-processor-1"
   └─ enable.idempotence = true (required)

2. Producer calls initTransactions():
   ├─ Sends FindCoordinatorRequest
   ├─ Connects to coordinator
   ├─ Sends InitProducerIdRequest
   └─ Receives PID and epoch

3. Producer calls beginTransaction():
   └─ Local state change (no broker call yet)

4. Producer sends records:
   ├─ Records include PID, epoch, sequence number
   ├─ Broker writes to partition log
   └─ Broker updates transaction log (add partition to transaction)

5. Transaction now "Ongoing" in coordinator state

Committing a Transaction

COMMIT TRANSACTION:

1. Producer calls commitTransaction():
   └─ Sends EndTxnRequest(commit) to coordinator

2. Coordinator:
   a. Transition state: Ongoing → PrepareCommit
   b. Write state to transaction log (durable)

3. Coordinator sends WriteTxnMarkersRequest to all partitions involved:
   └─ "Write COMMIT marker for PID=12345, epoch=0"

4. Partition leaders:
   ├─ Write COMMIT control record to log
   ├─ Respond to coordinator: Success
   └─ Consumer isolation level determines visibility

5. Coordinator:
   ├─ Transition state: PrepareCommit → CompleteCommit
   ├─ Write to transaction log
   └─ Respond to producer: Commit successful

6. Producer ready for next transaction

Aborting a Transaction

ABORT TRANSACTION:

1. Producer calls abortTransaction() OR transaction times out

2. Coordinator:
   a. Transition state: Ongoing → PrepareAbort
   b. Write state to transaction log

3. Coordinator sends WriteTxnMarkersRequest to all partitions:
   └─ "Write ABORT marker for PID=12345, epoch=0"

4. Partition leaders:
   ├─ Write ABORT control record to log
   ├─ Marks all records in transaction as aborted
   └─ Respond to coordinator: Success

5. Coordinator:
   ├─ Transition state: PrepareAbort → CompleteAbort
   ├─ Write to transaction log
   └─ Respond to producer: Abort successful

6. Aborted records remain in log but invisible to read_committed consumers

6. Transaction Markers

Control Records

COMMIT/ABORT MARKERS:

Control records:
└─ Special records in partition log
└─ Not visible to consumers (filtered out)
└─ Indicate transaction boundaries

COMMIT marker:
├─ Type: CONTROL_RECORD
├─ ControlType: COMMIT
├─ PID: 12345
├─ Epoch: 0
└─ Offset: 1050

ABORT marker:
├─ Type: CONTROL_RECORD
├─ ControlType: ABORT
├─ PID: 12345
├─ Epoch: 0
└─ Offset: 1055

Purpose:
└─ Mark end of transaction
└─ Enable read_committed isolation

LSO (Last Stable Offset)

LAST STABLE OFFSET:

Definition:
└─ Offset of first message in open transaction
└─ Or LEO if no open transactions

Calculation:
├─ Scan log for ongoing transactions
├─ Find earliest offset of open transaction
└─ LSO = that offset (or LEO if none)

Example:
├─ Offsets 0-999: Committed
├─ Offsets 1000-1010: Transaction 1 (ongoing)
├─ Offsets 1011-1020: Committed
└─ LSO: 1000 (first open transaction)

read_committed consumers:
└─ Can only read up to LSO
└─ Ensures they never see uncommitted data

7. Exactly-Once Semantics (EOS)

Broker-Side Guarantees

BROKER ENSURES:

1. Idempotent Writes:
   ├─ Producer sends: PID, epoch, sequence number
   ├─ Broker deduplicates based on sequence
   └─ Retries don't create duplicates

2. Transactional Writes:
   ├─ Multiple writes across partitions
   ├─ Atomic commit (all visible or none)
   └─ No partial writes visible

3. Fencing:
   ├─ Epoch increments fence old producers
   └─ Prevents zombie writes

4. Coordinator Durability:
   ├─ Transaction state written to replicated log
   └─ Survives coordinator failure

Combined:
└─ Exactly-once semantics (end-to-end with EOS pipeline)

Isolation Levels

CONSUMER ISOLATION LEVELS:

read_uncommitted (default):
└─ Consumers see all records (including uncommitted)
└─ High throughput, no transaction semantics

read_committed:
├─ Consumers only see committed records
├─ Reads up to LSO (Last Stable Offset)
├─ Aborted records filtered out
└─ Ensures transactional consistency

Configuration:
└─ isolation.level = read_committed (consumer config)

Trade-off:
├─ read_committed: Slight latency (wait for commit markers)
└─ read_uncommitted: No transaction guarantees

8. Coordinator Failover

Transaction Recovery

COORDINATOR FAILS:

1. Transaction coordinator (Broker 2) dies

2. Controller detects failure

3. New leader elected for __transaction_state partition
   └─ New coordinator: Broker 3

4. New coordinator loads transaction log:
   ├─ Reads all transaction state
   ├─ Rebuilds in-memory state
   └─ Identifies incomplete transactions

5. For each incomplete transaction:
   a. If PrepareCommit: Complete commit (write markers)
   b. If PrepareAbort: Complete abort (write markers)
   c. If Ongoing: Timeout (abort)

6. New coordinator fully operational

Recovery time:
└─ Seconds (depends on transaction log size)

9. Transaction Performance

Costs

TRANSACTION OVERHEAD:

1. Coordinator Communication:
   └─ InitProducerIdRequest, EndTxnRequest

2. Transaction Log Writes:
   └─ State changes written to replicated log

3. Marker Writes:
   └─ COMMIT/ABORT markers to all involved partitions

4. LSO Calculation:
   └─ Broker scans for open transactions

5. Read Latency (read_committed):
   └─ Consumers wait for commit markers

Impact:
├─ Throughput: ~10-20% reduction (vs non-transactional)
├─ Latency: +10-50ms (commit marker propagation)
└─ Acceptable for EOS guarantee

Optimizations

REDUCING OVERHEAD:

1. Batch transactions:
   └─ Large transactions amortize fixed costs

2. Minimize partitions per transaction:
   └─ Fewer marker writes

3. Tune transaction.timeout.ms:
   └─ Balance responsiveness vs abort rate

4. Use dedicated coordinator brokers:
   └─ Isolate coordinator load (not always needed)

5. Monitor transaction metrics:
   └─ transaction-count, transaction-duration

10. Key Takeaways

You've learned:

✅ Transaction coordinator manages transaction lifecycle
✅ Transaction log (__transaction_state) stores transaction state
✅ Producer ID (PID) + epoch identify and fence producers
✅ Transaction states: Ongoing → Prepare → Complete
✅ Commit markers indicate transaction boundaries
✅ LSO (Last Stable Offset) determines read_committed visibility
✅ Exactly-once semantics via idempotence, transactions, fencing
✅ Isolation levels: read_uncommitted vs read_committed
✅ Coordinator failover recovers incomplete transactions
✅ Transaction overhead ~10-20% throughput reduction (acceptable)

Summary

Kafka transactions enable exactly-once semantics:

Architecture =
├─ Transaction coordinator (broker-side role)
├─ Transaction log (__transaction_state, 50 partitions)
├─ Producer ID (PID) + epoch (fencing)
└─ Commit/abort markers (control records)

Transaction Flow =
├─ Begin: Producer gets PID, starts transaction
├─ Write: Records include PID, epoch, sequence
├─ Commit: Coordinator writes markers, transition state
└─ Abort: Coordinator writes abort markers

Guarantees =
├─ Idempotence: Sequence numbers deduplicate
├─ Atomicity: All writes visible or none
├─ Fencing: Epochs prevent zombie writes
└─ Durability: Transaction state replicated

Isolation =
├─ read_uncommitted: See all records (no EOS)
└─ read_committed: Only committed (EOS guarantee)

Coordinator Failover =
├─ New coordinator loads transaction log
├─ Completes incomplete transactions
└─ Seconds to recover

Remember: Kafka transactions are broker-coordinated and provide exactly-once semantics when used correctly. Understanding broker-side mechanics explains transaction behavior, performance characteristics, and failure recovery.

🎉 Congratulations!

You've completed all 15 modules of Kafka Internal Mechanics!

You now have deep knowledge of:

  • ✅ End-to-end message flow
  • ✅ Topics, partitions, and log structure
  • ✅ Replication and durability protocols
  • ✅ Leader election and failover
  • ✅ Cluster metadata and coordination (ZK and KRaft)
  • ✅ Controller internals
  • ✅ Broker lifecycle
  • ✅ Transactions and exactly-once semantics

This knowledge empowers you to:

  • Design highly available and durable Kafka architectures
  • Troubleshoot complex Kafka issues at the internals level
  • Optimize Kafka performance based on deep understanding
  • Confidently operate Kafka in production

Keep learning and experimenting with Kafka! 🚀