10.1.1 – Transaction Protocol and Coordinator

Module: M10 – Advanced Topics & Architecture
Section: 10.1 – Transactions & Exactly-Once Semantics
Goal: Deep dive into Kafka's transaction protocol and the transaction coordinator

Introduction

Kafka transactions enable exactly-once semantics (EOS) across producers and consumers, guaranteeing that messages are processed exactly once—no duplicates, no data loss—even in the face of failures, retries, or crashes.

This module is the authoritative source for understanding Kafka's transaction protocol. Other modules (Producers, Streams) reference this content but don't re-teach it.

What You'll Learn:

• The transaction coordinator's role in Kafka
• Transaction logs and control markers
• How the transaction protocol works end-to-end
• The difference between idempotent producers and transactional producers

Why Transactions Matter

The Problem: At-Least-Once Isn't Enough

At-least-once delivery (default in Kafka):

Producer sends message → Ack timeout → Producer retries → Duplicate message

Result: Messages may be duplicated if retries occur after the broker has already committed the message.

Real-World Impact:

Example: Payment processing system.

Order Service → Kafka → Payment Service

If Payment Service crashes after processing but before committing offset:
  1. Consumer restarts
  2. Re-reads message
  3. Processes payment AGAIN
  4. Customer charged twice ❌

Solution: Exactly-once semantics ensure idempotent processing—payments are never duplicated, even with retries or crashes.

Idempotent Producers vs Transactional Producers

Idempotent Producers

Configuration:

enable.idempotence=true  # Default in Kafka 3.0+

What It Does:

• Assigns a Producer ID (PID) and sequence number to each message
• Broker deduplicates messages with the same PID + sequence number
• Guarantees no duplicates within a single producer session

Guarantees:

• ✅ No duplicates per partition (within producer session)
• ❌ No atomicity across partitions
• ❌ No coordination with consumer offsets

Idempotent Producer Flow:

Producer (PID=123)
  ↓
Send: (key=order-1, value="paid", sequence=0)
  ↓
Broker receives → checks if sequence=0 already exists for PID=123
  ↓
If exists: Drop (duplicate)
If new: Append to log

Limitation: If producer crashes and gets a new PID, duplicates can occur across sessions.

Transactional Producers

Configuration:

enable.idempotence=true          # Required
transactional.id=payment-processor-1  # Unique per producer

What It Does:

• Extends idempotence to multiple partitions and consumer offsets
• Uses a transaction coordinator to ensure atomicity
• Guarantees all-or-nothing writes (commit or abort)

Guarantees:

• ✅ No duplicates across producer sessions (survives crashes)
• ✅ Atomic writes across multiple partitions
• ✅ Atomic offset commits (producer + consumer in one transaction)

When to Use Each:

Transaction Coordinator

What Is the Transaction Coordinator?

The transaction coordinator is a Kafka broker component responsible for managing the lifecycle of transactions.

Key Responsibilities:

1. Assign Producer IDs: Provides unique, persistent IDs to transactional producers
2. Track Transaction State: Maintains the state of ongoing transactions (begin, prepare, commit, abort)
3. Write Transaction Markers: Writes control markers to partition logs (commit/abort)
4. Ensure Atomicity: Guarantees all-or-nothing semantics across partitions

Transaction Coordinator vs Group Coordinator

Key Point: These are separate but often co-located on the same broker.

Transaction Coordinator Election

How Is the Coordinator Chosen?

The transaction coordinator for a given transactional.id is determined by:

Coordinator Broker = hash(transactional.id) % num_partitions(__transaction_state)

Example:

transactional.id = "payment-processor-1"
__transaction_state topic has 50 partitions

Coordinator = hash("payment-processor-1") % 50 = partition 23
Broker hosting partition 23 becomes the coordinator

Fault Tolerance: If the coordinator broker fails, another broker hosting a replica of the transaction log partition becomes the new coordinator.

Transaction Log: __transaction_state

What Is the Transaction Log?

The __transaction_state topic is an internal Kafka topic that stores the state of all active and completed transactions.

Key Characteristics:

• Compacted: Old transaction states are compacted (keeps only the latest state per transactional.id)
• Replicated: Typically replication.factor=3 for fault tolerance
• 50 partitions (default): Distributes transaction state across brokers

Critical Broker Configurations for Transaction State

The __transaction_state topic must be properly configured for transaction reliability:

Example server.properties Configuration:

# Transaction state log replication
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2

# Transaction timeouts
transaction.timeout.ms=60000
transaction.max.timeout.ms=900000

# Transaction state log partitions
transaction.state.log.num.partitions=50

Why These Matter:

1. transaction.state.log.replication.factor

• If this topic is under-replicated and a broker fails, transaction state can be lost
• Critical: Set to 3 in production to survive broker failures
• If set too low, transaction coordinator failover can cause data loss

2. transaction.state.log.min.isr

• Ensures transaction state is written to at least N replicas before acknowledging
• Critical: Set to 2 to ensure coordinator failover can recover state
• If set to 1, broker failure before replication completes = lost transaction state

3. transaction.timeout.ms

• Transactions open longer than this are automatically aborted by the coordinator
• Tuning: Increase for slow processing pipelines (batch jobs)
• Warning: Very long timeouts increase risk of resource exhaustion

Real-World Impact Example:

Scenario: transaction.state.log.replication.factor = 1

1. Producer starts transaction
2. Coordinator writes state to __transaction_state (1 replica)
3. Coordinator broker crashes before replication
4. New coordinator has no record of transaction
5. Transaction is lost (partial writes remain uncommitted forever)

Result: Consumer blocked on read_committed (waiting for marker that never comes)

Monitoring Transaction State Log Health:

# Check replication status
kafka-topics.sh --describe \
  --topic __transaction_state \
  --bootstrap-server localhost:9092

# Look for:
# - ReplicationFactor: 3 ✅
# - Isr: [1,2,3] (all replicas in sync) ✅
# - Isr: [1] (under-replicated) ❌

Transaction Log Entry Structure

Entry Format:

Key: transactional.id
Value: {
  pid: 12345,                     // Producer ID
  epoch: 1,                       // Producer epoch (increments on restart)
  status: "Ongoing",              // Ongoing, PrepareCommit, PrepareAbort, CompleteCommit, CompleteAbort
  partitions: [                   // Partitions involved in transaction
    {topic: "orders", partition: 0},
    {topic: "payments", partition: 2}
  ],
  timeout_ms: 60000              // Transaction timeout
}

Transaction State Transitions:

Empty (no transaction)
  ↓
Ongoing (transaction started, writes happening)
  ↓
PrepareCommit / PrepareAbort (producer requests commit/abort)
  ↓
CompleteCommit / CompleteAbort (markers written to all partitions)
  ↓
(state removed after compaction)

Transaction Protocol: Step-by-Step

Step 1: Producer Initialization

Producer sends InitProducerIdRequest to transaction coordinator:

Properties props = new Properties();
props.put("transactional.id", "payment-processor-1");
props.put("enable.idempotence", true);

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();  // Sends InitProducerIdRequest

What Happens:

1. Transaction coordinator checks __transaction_state for existing PID
2. If exists: Increment epoch (new producer session)
3. If new: Assign new PID
4. Return {pid: 12345, epoch: 1} to producer

Result: Producer has a persistent PID + epoch, survives restarts.

Step 2: Begin Transaction

Producer calls beginTransaction():

producer.beginTransaction();

What Happens:

• Producer tracks transaction state locally (no network call yet)
• Transaction coordinator not notified until first message sent

Step 3: Produce Messages

Producer sends messages to multiple partitions:

producer.send(new ProducerRecord<>("orders", "order-1", "paid"));
producer.send(new ProducerRecord<>("payments", "payment-1", "processed"));

What Happens:

1. Producer sends AddPartitionsToTxnRequest to coordinator (first message only)
2. Coordinator writes to __transaction_state: "Transaction Ongoing, partitions: [orders-0, payments-2]"
3. Producer sends messages to partition leaders
4. Messages written with transaction metadata (PID, epoch, sequence number)
5. Messages marked as uncommitted (not visible to read_committed consumers yet)

Step 4: Send Offsets to Transaction (Optional)

If this is a read-process-write pipeline:

Map<TopicPartition, OffsetAndMetadata> offsets = ...;
producer.sendOffsetsToTransaction(offsets, "consumer-group-id");

What Happens:

1. Producer sends AddOffsetsToTxnRequest to coordinator
2. Coordinator adds __consumer_offsets partition to transaction
3. Producer sends offset commit to group coordinator as part of the transaction

Result: Consumer offsets are committed atomically with produced messages.

Step 5: Commit Transaction

Producer calls commitTransaction():

producer.commitTransaction();

What Happens (Two-Phase Commit):

Phase 1: Prepare Commit

1. Producer sends EndTxnRequest(commit=true) to transaction coordinator
2. Coordinator writes "PrepareCommit" to __transaction_state
3. Coordinator sends WriteTxnMarkersRequest to all partition leaders involved

Phase 2: Write Markers 4. Each partition leader writes a COMMIT control marker to its log 5. Partition leaders respond to coordinator 6. Coordinator writes "CompleteCommit" to __transaction_state 7. Coordinator responds to producer with success

Result: All messages in the transaction are now visible to read_committed consumers.

Step 6: Abort Transaction (Failure Path)

If an error occurs:

try {
    producer.beginTransaction();
    producer.send(...);
    // Error occurs
    producer.abortTransaction();  // Rollback
} catch (Exception e) {
    producer.abortTransaction();
}

What Happens:

1. Producer sends EndTxnRequest(commit=false) to coordinator
2. Coordinator writes "PrepareAbort" to __transaction_state
3. Coordinator sends abort markers to all partitions
4. Partition leaders write ABORT control marker to logs
5. Coordinator writes "CompleteAbort" to __transaction_state

Result: All messages in the transaction are invisible to read_committed consumers (treated as never written).

Control Markers

What Are Control Markers?

Control markers are special records written to partition logs to signal the end of a transaction.

Types:

• COMMIT marker: Makes all messages in the transaction visible
• ABORT marker: Makes all messages in the transaction invisible

Control Marker Structure:

Offset: 1000
Key: {
  version: 0,
  coordinatorEpoch: 5
}
Value: {
  type: COMMIT (or ABORT),
  pid: 12345,
  producerEpoch: 1
}

How Consumers Use Markers:

read_committed Consumer:

Partition Log:
  Offset 100: message (pid=12345, uncommitted)
  Offset 101: message (pid=12345, uncommitted)
  Offset 102: COMMIT marker (pid=12345)
  Offset 103: message (pid=67890, uncommitted)

Consumer with read_committed:
  → Reads offsets 100, 101 (visible after seeing COMMIT marker at 102)
  → Blocks on offset 103 (waits for marker)

read_uncommitted Consumer:

  → Reads all messages immediately (ignores markers)

Transaction Timeout

Configuration:

transaction.timeout.ms=60000  # Default: 1 minute

What Happens on Timeout?

If a transaction doesn't commit within transaction.timeout.ms:

1. Transaction coordinator aborts the transaction automatically
2. Writes ABORT markers to all partitions
3. Producer receives ProducerFencedException on next operation

Why This Matters: Prevents transactions from holding resources indefinitely (e.g., if producer crashes mid-transaction).

Tuning Guidance:

Warning: Very long timeouts increase risk of resource exhaustion if producers crash.

Producer Epochs and Fencing

What Is Producer Epoch?

Producer epoch = A monotonically increasing counter that increments every time a producer with the same transactional.id restarts.

Purpose: Prevents "zombie" producers (old instances that haven't realized they've been replaced) from writing data.

Fencing Example:

Producer A (PID=12345, epoch=1) starts
Producer A crashes (doesn't clean up)

Producer B (same transactional.id) starts
  → Coordinator increments epoch to 2
  → Returns PID=12345, epoch=2

Producer A comes back online (still thinks epoch=1)
  → Sends message with epoch=1
  → Broker rejects: "Your epoch is outdated, you've been fenced"

Producer B continues with epoch=2 ✅

Key Point: Only the producer with the highest epoch can write. Old instances are automatically fenced.

Transaction Coordinator Failure

What Happens If Coordinator Fails?

1. Broker hosting coordinator crashes
2. Follower replica of __transaction_state partition becomes leader
3. New broker becomes transaction coordinator
4. Ongoing transactions are aborted (coordinator doesn't know their state)
5. Producer retries InitProducerIdRequest with new coordinator
6. New transaction starts with incremented epoch

Result: Transactions are aborted, but data consistency is maintained (no partial writes).

Key Takeaways

1. Transaction Coordinator: Manages transaction lifecycle, assigns PIDs, writes markers.

2. Transaction Log (__transaction_state): Stores state of all transactions, replicated for fault tolerance.

3. Idempotent vs Transactional: Idempotent prevents duplicates per session; transactional provides atomicity across partitions and offsets.

4. Two-Phase Commit: Coordinator prepares commit/abort, writes markers to all partitions, then completes.

5. Control Markers: COMMIT/ABORT markers in logs signal transaction outcome to consumers.

6. Producer Epochs: Prevent zombie producers via fencing (only highest epoch can write).

7. Transaction Timeout: Automatically aborts transactions that don't complete in time.

What's Next?

Next Module: 10.1.2 – Producer Transactions

What We Covered:

• ✅ Transaction protocol architecture
• ✅ Transaction coordinator role
• ✅ Transaction log structure
• ✅ Control markers
• ✅ Producer epochs and fencing

What's Coming:

• ❌ Producer transactional API (code examples)
• ❌ Consumer isolation levels
• ❌ End-to-end EOS pipelines
• ❌ EOS anti-patterns and debugging

References to Other Modules:

• Producer retry configs → M3: Producers
• Replication mechanics → M7: Kafka Internals
• EOS in Streams → M5: Kafka Streams