Kafka Exactly-Once Semantics (EOS): Idempotent Producers & 2PC Transactions

Suppose you are building a banking stream processor.

Your microservice consumes financial transfer events from an input topic (account-transfers), applies currency conversion logic, publishes transformed transactions to an output topic (processed-transfers), and commits its consumer read offset.

Now consider what happens if a server crash occurs midway through execution:

• If the service publishes to processed-transfers but crashes before committing its input offset, the restarted service re-reads the input message and processes the transaction a second time.
• If the service commits its input offset but crashes before publishing output records, the output transfer vanishes.

Achieving Exactly-Once Semantics (EOS) across read-process-write loops in a distributed system was long considered impossible.

Kafka achieved EOS by combining Idempotent Producers with Distributed Transactions.

---

The Two Guarantees That Form EOS

Exactly-Once Semantics is not a single setting. It is the combination of two underlying capabilities:

$\text{Exactly-Once Semantics (EOS)} = \text{Idempotent Producer} + \text{Transactional Coordinator (2PC)}$

---

Element 1: The Idempotent Producer

As we saw in Part 08, network retries can cause duplicate messages on brokers if an ACK packet is lost over the network.

When you enable enable.idempotence = true:

1. The broker assigns a unique 64-bit Producer ID (PID) to the producer client during initialization.
2. The producer assigns a monotonically increasing Sequence Number (0, 1, 2…) to every record batch per partition.
3. The broker tracks the highest sequence number written for each PID.

Broker Receives Batch (PID: 402, Sequence: 5) ---> Writes to Log & Updates LastSeq = 5
Network drops ACK. Producer retries Batch (PID: 402, Sequence: 5).
Broker receives duplicate! LastSeq is ALREADY 5.
Broker returns ACK to producer but DISCARDS duplicate bytes!

Idempotence eliminates duplicate messages caused by producer network retries on a single partition.

---

Element 2: Transactional Read-Process-Write Loops

When a stream processing application reads from input topic A and writes to output topic B, it must update output records AND input consumer offsets atomically.

Either both writes succeed, or both writes are aborted.

                                STREAM PROCESSOR
                             +-------------------+
                             | read input msg    |
                             | transform data    |
                             | publish output    |
                             | commit offset     |
                             +-------------------+
                                /             \
                               v               v
               TOPIC B (Output Records)     __consumer_offsets (Input Offsets)
               ========================     =================================
               [ MUST SUCCEED TOGETHER OR ABORT TOGETHER IN A SINGLE TRANSACTION ]

---

The Transactional Coordinator & Two-Phase Commit (2PC)

Kafka manages cross-topic atomic writes using a Transactional Coordinator broker and an internal topic named __transaction_state.

Step-by-Step Transaction Execution Flow

Here is the exact sequence executed by a transactional application:

// 1. Configure Producer Transactional ID
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "banking-processor-1");
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

// 2. Initialize Transactions (Fences out old zombie producer instances!)
producer.initTransactions();

try {
    // 3. Begin Transaction Block
    producer.beginTransaction();

    // 4. Publish output messages to destination topic
    producer.send(new ProducerRecord<>("processed-transfers", key, value));

    // 5. Send input consumer offsets to the TRANSACTION (not to __consumer_offsets directly!)
    producer.sendOffsetsToTransaction(currentOffsets, consumerGroupId);

    // 6. Commit Two-Phase Transaction
    producer.commitTransaction();
} catch (ProducerFencedException e) {
    producer.close();
} catch (KafkaException e) {
    producer.abortTransaction(); // Aborts transaction on failure!
}

---

Under the Hood: Two-Phase Commit Protocol

When producer.commitTransaction() is called, Kafka executes a Two-Phase Commit (2PC):

PHASE 1: PREPARE COMMIT
1. Producer requests commit from Transactional Coordinator.
2. Coordinator writes "PREPARE_COMMIT" marker to __transaction_state topic.

PHASE 2: WRITE CONTROL MARKERS
3. Coordinator writes a special COMMIT CONTROL MARKER to output topic partition.
4. Coordinator writes a COMMIT CONTROL MARKER to __consumer_offsets partition.
5. Coordinator writes "COMPLETE_COMMIT" marker to __transaction_state topic.

OUTPUT TOPIC PARTITION LOG FILE (.log)
[ Offset 100: Record A ] -> [ Offset 101: Record B ] -> [ CONTROL MARKER: COMMIT ]

---

Consumer Isolation Levels: read_committed vs read_uncommitted

How do downstream consumers handle transaction control markers and uncommitted messages?

Consumers specify their isolation.level configuration:

1. isolation.level = read_uncommitted (Default)

The consumer reads all records in the log, including uncommitted messages and messages from aborted transactions.

2. isolation.level = read_committed

The consumer buffers messages internally and suppresses uncommitted records. It only delivers records up to the Last Stable Offset (LSO) where transactions have been explicitly committed.

LOG FILE AT ISOLATION LEVEL read_committed:

Offset 10: [ Committed Msg ]  ===> Delivered to Consumer
Offset 11: [ Aborted Msg ]    ===> Skipped silently by Consumer!
Offset 12: [ Control Marker ] ===> Used for internal tracking (never delivered)

Messages belonging to aborted transactions are filtered out automatically by the consumer client SDK.

---

Quick Summary

• Exactly-Once Semantics (EOS) combines Idempotent Producers (PID + Sequence Numbers) with Transactional Coordinators.
• sendOffsetsToTransaction() binds input consumer offset commits to output message writes within a single atomic transaction.
• The Transactional Coordinator executes a Two-Phase Commit (2PC), writing COMMIT or ABORT control markers into partition logs.
• Downstream consumers setting isolation.level = read_committed filter out aborted messages automatically.

References & Further Reading

1. Apache Software Foundation. Apache Kafka Documentation: Kafka Connect Developer Guide. Apache Kafka Docs.
2. Confluent Inc. Designing Custom Connectors for Kafka Connect. Confluent Docs.
3. Hohpe, G., & Woolf, B. (2003). Enterprise Integration Patterns. Addison-Wesley.

---