Exactly-Once #

In the world of distributed software engineering, guaranteeing that a message is processed exactly once is often considered the “Holy Grail” that’s impossible to achieve. Based on distributed system academic theory (like the FLP Impossibility theorem), it’s very hard to distinguish between temporary network failures and permanent crashes on a node without expensive centralized coordination. However, in June 2017, Apache Kafka released version 0.11 which introduced the Exactly-Once Semantics (EOS) feature. This guarantee ensures that even if network failures, connection disruptions, broker crashes, or sudden application restarts happen mid-processing, the data we send from producers to brokers and process with consumers is recorded exactly once consistently. This exactly-once processing is critical for sensitive applications like financial transaction processing, warehouse inventory systems, and logistics calculations.

Theoretical Foundation: Why Is Exactly-Once So Hard to Achieve? #

To understand the genius of the EOS solution in Kafka, we must understand why this delivery model has historically been so hard to achieve. The main problem lies in Network State Uncertainty. When a sender (producer) sends a message to a receiver (broker) and the network disconnects before the sender receives a reply, the sender faces two unknown possibilities:

  1. The message never arrived at the broker (data loss if not resent).
  2. The message was already safely stored by the broker, but the return ACK packet was lost in transit (duplication if resent).

In traditional systems, to solve this problem, developers had to choose between accepting data loss (At-Most-Once) or accepting duplicate data risk (At-Least-Once) and then filtering it manually using complex database-level deduplication logic.

Kafka solves this problem by unifying three internal architecture pillars that collaborate to form a cohesive Exactly-Once system:

flowchart TD
    EOS["Exactly-Once Semantics (EOS)"]
    
    EOS --> P1["1. Idempotent Producer"]
    EOS --> P2["2. Transactional API"]
    EOS --> P3["3. Isolation Level (read_committed)"]
    
    P1 -->|"Prevents Network Duplication"| PID["PID Identity + Sequence Number"]
    P2 -->|"Atomic Multi-Partition Writes"| TC["Transaction Coordinator + Transaction State Log"]
    P3 -->|"Consumer Filtering"| LSO["Last Stable Offset (LSO) + Commit Marker"]

    style EOS stroke:#0288d1,stroke-width:2px
    style P1 stroke:#2e7d32,stroke-width:2px
    style P2 stroke:#2e7d32,stroke-width:2px
    style P3 stroke:#2e7d32,stroke-width:2px

The three components above work sequentially. The Idempotent Producer guarantees no duplicate messages are written by the producer to one partition from network resends. The Transactional API guarantees writes to multiple partitions and different topics (including consumer offsets) run atomically (all succeed or all abort). Finally, the Isolation Level on the consumer side guarantees downstream applications don’t read data from failed or ongoing transactions.

Pillar 1: Idempotent Producer #

Producer idempotence is the foundation of EOS. Without idempotence, it’s impossible to build safe transactions. Idempotence guarantees that no matter how many times the producer resends the same message because of ACK loss on the network, the broker only writes that message exactly once to the physical disk log.

How Idempotence Works Internally #

When idempotence is enabled (enable.idempotence=true), Kafka does two things behind the scenes:

1. Producer ID (PID) Allocation #

Every time a producer is initialized, it requests a unique ID from the coordinator broker using the InitProducerId API call. The broker gives a Producer ID (PID) and increments the producer’s Epoch value. This PID is unique within the Kafka cluster and acts as that producer’s unique identity.

2. Sequence Numbers #

For each destination partition, the producer assigns a Sequence Number starting from 0 and incrementing by 1 for every message sent. This sequence number is embedded in the binary message header along with the PID.

On the broker side, active broker memory maintains a state map for each (PID, Partition) pair recording the last successfully written sequence number (Last Sequence Number). When the broker receives a new message, it checks that message’s sequence number:

  • If Sequence Number == Last Sequence + 1: The message is valid, the broker writes it to disk and updates Last Sequence to the new sequence number.
  • If Sequence Number <= Last Sequence: The broker realizes this message is a duplicate from producer resends. The broker silently discards that message (doesn’t write to disk) but still sends a success ACK to the producer so the producer doesn’t worry and doesn’t keep trying to resend.
  • If Sequence Number > Last Sequence + 1: Indicates a missed message (out-of-sequence), which means data loss happened on the network before the message arrived. The broker rejects that message and sends the OutOfOrderSequenceException error to the producer.
sequenceDiagram
    autonumber
    participant Prod as Producer (PID=100)
    participant Broker as Kafka Broker (State: PID 100 -> LastSeq 5)

    Note over Prod, Broker: "ProduceRequest with Seq 6 (Normal)"
    Prod->>Broker: ProduceRequest (PID=100, Seq=6, Data A)
    Broker->>Broker: Check Seq: 6 == LastSeq 5 + 1 (Valid!)
    Broker->>Broker: Write Data A to Disk
    Note over Broker: Update State: PID 100 -> LastSeq 6
    Broker-->>Prod: Success ACK

    Note over Prod, Broker: "ProduceRequest with Seq 6 (Duplicate from Retry)"
    Prod->>Broker: ProduceRequest (PID=100, Seq=6, Data A)
    Broker->>Broker: Check Seq: 6 <= LastSeq 6 (DUPLICATE!)
    Note over Broker: Message silently discarded from disk writes
    Broker-->>Prod: Success ACK (Client considers delivery successful)

Producer Idempotence Limitations #

Although very powerful, producer idempotence has two important limitations:

  1. Bound to the Producer Session (Single-Session): If the producer application fully restarts, it’s allocated a new PID by the broker. The broker considers this a new producer, so duplicates from the old producer session stuck in the network queue can no longer be filtered.
  2. Bound to a Single Partition: Idempotence only guarantees no duplication on one partition of one topic. It can’t guarantee atomic transactions across partitions or topics. To overcome this limitation, we need the Transactional API.

Pillar 2: Transactional API #

To overcome idempotence limitations, Kafka provides the Transactional API. This allows producers to send a group of messages to various partitions and topics atomically. This atomicity guarantee follows the All-or-Nothing principle: all messages in a transaction are successfully committed, or no messages are visible to consumers if the transaction is aborted.

The Kafka transaction protocol is coordinated by an internal broker component called the Transaction Coordinator. This Coordinator tracks transaction status in a safe internal log topic named __transaction_state.

The Kafka transaction lifecycle involves the following steps:

  1. Initialization: The producer registers a static transactional.id with the Transaction Coordinator. This provides Zombie Fencing guarantees (fencing off old producers if two producer instances with the same transaction ID run simultaneously).
  2. Starting the Transaction: The producer calls beginTransaction() to start a new transaction block locally.
  3. Sending Messages: The producer sends messages to various topic partitions. Behind the scenes, the Transaction Coordinator records which partitions are involved in this transaction into the __transaction_state topic.
  4. Writing the Commit/Abort Marker: When the producer calls commitTransaction(), the Transaction Coordinator writes a PrepareCommit closing record to __transaction_state, then spreads a special message called the Commit Marker (or Abort Marker if the transaction is aborted) to all involved physical topic partitions.
PARTITION LOG CONTENT WITH COMMIT MARKER:
┌─────────────────────┬─────────────────────┬─────────────────────┐
│ Offset 200: Msg A   │ Offset 201: Msg B   │ Offset 202: COMMIT  │
│ (Tx Active)         │ (Tx Active)         │ (Commit Marker)     │
└─────────────────────┴─────────────────────┴─────────────────────┘
                                                    │
                                         (read_committed consumer)
                                                    │
                                                    ▼
                                     (Only reads data after Offset 202)

Pillar 3: The read_committed Isolation Level on the Consumer Side #

The atomic write guarantee on the broker is useless if consumers immediately process every incoming data row without checking that data’s transaction status. Therefore, on the downstream consumer side, we must configure the property:

  • isolation.level=read_committed

By default, this property is read_uncommitted, meaning consumers immediately devour all data in the partition log without caring whether that data comes from successful, ongoing, or aborted transactions.

How Does read_committed Work? #

When a consumer is set to read_committed, the consumer client library does client-side filtering using information from the broker:

  1. The consumer reads the partition log sequentially.
  2. If it finds a transactional message, it checks whether that message is followed by a Commit Marker at the next offset.
  3. If it finds a Commit Marker, the message is decoded and forwarded to our application’s business logic loop.
  4. If it finds an Abort Marker, the consumer automatically silently skips messages from that aborted transaction. Our application never sees those message contents.
  5. Holding the LSO (Last Stable Offset): If there’s an ongoing transaction on that partition, the consumer isn’t allowed to read any message beyond the LSO, even if new non-transactional messages were written after that ongoing transaction started. This ensures consumers don’t skip over uncertain transaction states.

Java SDK Implementation Code: The Best EOS Configuration #

Here’s a Java code example comparing non-idempotent producer writing (anti-pattern for EOS) with a safe idempotent producer for production environments.

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;

public class ExactlyOnceProducerComparison {
    public static void main(String[] args) {
        Properties baseProps = new Properties();
        baseProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        baseProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        baseProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // =====================================================================
        // 1. ANTI-PATTERN: NON-IDEMPOTENT PRODUCER (Duplication Prone)
        // =====================================================================
        Properties unsafeProps = new Properties(baseProps);
        unsafeProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false");
        unsafeProps.put(ProducerConfig.ACKS_CONFIG, "1"); // Only waits for the leader
        unsafeProps.put(ProducerConfig.RETRIES_CONFIG, 3);
        
        KafkaProducer<String, String> unsafeProducer = new KafkaProducer<>(unsafeProps);
        // DON'T use the configuration above if you're processing balance payments!
        unsafeProducer.send(new ProducerRecord<>("payments", "user_10", "{\"amount\": 50000}"));
        unsafeProducer.close();


        // =====================================================================
        // 2. CORRECT: IDEMPOTENT PRODUCER (Safe From Network Duplication)
        // =====================================================================
        Properties safeProps = new Properties(baseProps);
        
        // ✓ Enabling the idempotence guarantee at the single-partition level
        safeProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
        
        // Important Note: When enable.idempotence is set to true, the Kafka client
        // automatically validates and forces the following parameters:
        // - ACKS_CONFIG is forced to "all" (maximum durability)
        // - RETRIES_CONFIG is forced to Integer.MAX_VALUE
        // - MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION is forced to max 5 (keeping message order)
        
        KafkaProducer<String, String> safeProducer = new KafkaProducer<>(safeProps);
        
        // Safe from duplication risk from momentary network switch outages
        safeProducer.send(new ProducerRecord<>("payments", "user_10", "{\"amount\": 50000}"), new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception != null) {
                    // A fatal non-retriable error occurred (e.g., ACL authorization failure)
                    System.err.println("✗ Failed to send the transaction idempotently: " + exception.getMessage());
                } else {
                    System.out.println("✓ Transaction successfully stored at offset: " + metadata.offset());
                }
            }
        });
        
        safeProducer.close();
    }
}

When to Choose Exactly-Once Semantics? #

To help system architects choose the right data delivery guarantee level, use the following practical guide:

STILL use Exactly-Once if:
  ✓ The application processes financial data, accounting ledgers, billing, or money transfers.
  ✓ You're doing sensitive data aggregation (like calculating total inventory in real-time).
  ✓ You use Kafka Streams for stateful processing (windowing join/aggregation) across topics.
  ✓ Message duplication absolutely cannot be tolerated by the target database.

DON'T force Exactly-Once (Use At-Least-Once) if:
  ✗ Ultra-high throughput (e.g., processing >10,000,000 IoT metrics per second) is the main goal.
  ✗ Consistent end-to-end latency under 5 milliseconds is a must (transaction overhead hinders this).
  ✗ Your downstream database system is already idempotent (more efficient to use DB-side deduplication).

Summary #

  • Exactly-Once Semantics (EOS) — The strongest delivery guarantee where messages are confirmed to enter the broker log and finish executing on consumers exactly once, free from data loss and duplication.
  • Idempotent Producer — The foundational EOS component preventing message duplication from network retries by embedding Producer IDs (PID) and Sequence Numbers in data packet headers.
  • Sequence Number Filtering — The broker mechanism for silently discarding duplicate packets if the received message sequence number is less than or equal to the last sequence number recorded in broker memory.
  • Transactional API — A set of API functions guaranteeing cross-topic and cross-partition write atomicity (All-or-Nothing) coordinated by the Transaction Coordinator through the __transaction_state topic.
  • Isolation Level read_committed — The mandatory consumer-side setting to actively filter and discard data from aborted transactions and hold reads at the Last Stable Offset (LSO) boundary.
  • Zombie Fencing — The security protocol for isolating and rejecting write access from old producer instances (split-brain) after new producers with the same transaction ID register to the broker.
  • Performance Overhead — EOS transactions cause a slight throughput decrease and latency increase from the extra commit marker writes and LSO holding on the broker.

← Previous: At-Least-Once Next: Transaction API →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact