Exactly-Once Semantics #

In real-time data processing architecture, data accuracy is the key. When we use asynchronous message delivery models, mid-stream infrastructure failures often force our systems to reprocess data, triggering data duplication (the At-Least-Once model) or even data loss (the At-Most-Once model). To achieve the highest reliability level, Apache Kafka introduced the Exactly-Once Semantics (EOS) guarantee. EOS isn’t just a single feature; it’s a complex architectural integration between the Idempotent Producer feature and the Transaction API on the producer side, collaborating with consumer isolation level tuning. The EOS guarantee ensures that within the data processing flow, every input message is processed, internal state is updated, and output messages are written to the destination topic atomically — everything happens exactly once, no less and no more, even if the broker cluster or our application crashes mid-process.

Basic Concept of Exactly-Once Semantics (EOS) #

To understand why EOS is so revolutionary, we must look at the common context where this guarantee is most needed: the read-process-write processing pattern.

This pattern is the foundation of stream processing architectures (like Kafka Streams or custom microservices applications):

  1. Read: A consumer reads messages from Input Topic A (e.g., PaymentRequests).
  2. Process: The application processes that data (e.g., validates balance and deducts admin fees).
  3. Write: The producer writes the calculation result messages to Output Topic B (e.g., ProcessedTransactions) and simultaneously records the input message offsets to the internal __consumer_offsets topic.

Without a transactional system (EOS), if our application crashes right after step 3 (writing to Output Topic B) but before updating the offsets in step 4, then when the application restarts, it re-reads the same input messages and sends duplicate transaction messages to Output Topic B.

EOS solves this problem by treating the Write (Output) and Write (Offsets) steps as one atomic transaction unit. If one step fails, the entire transaction is aborted, and downstream consumers never see that half-baked data.

Kafka’s Transactional Architecture: The Role of the Transaction Coordinator #

Kafka’s distributed transaction system doesn’t rely on the slow, database-connection-blocking traditional two-phase commit (2PC) protocol. Kafka implements a log-based transactional architecture coordinated by an internal component called the Transaction Coordinator.

Here are the three main architectural components managing Kafka transactions:

1. Transaction Coordinator #

This is a dedicated Kafka broker managing transaction lifecycles from producers. Its job is similar to the Group Coordinator that manages consumers. The leader broker of the internal __transaction_state topic automatically acts as the Transaction Coordinator for the related producer.

2. Transaction Journal Topic (__transaction_state) #

All transaction status changes (Ongoing, PrepareCommit, CompleteCommit) are permanently recorded in a high-partition internal topic named __transaction_state. This topic is replicated across all cluster brokers to guarantee that if the active Transaction Coordinator dies, the replacement ISR broker can read that journal and continue coordinating hanging transactions.

3. The transactional.id Property #

For transactions to survive producer crash and restart scenarios, we must define a unique, persistent transactional.id property on the producer side (e.g., transactional.id=produser-transaksi-pembayaran-0).

  • This property acts as a permanent identity. When a newly started producer registers the same transactional.id, the broker recognizes its old session, aborts hanging uncommitted transactions from the old session (epoch fencing), and allows the new producer to start a clean transaction.

Transaction Mechanism: The read-process-write Flow #

Here are the 7 chronological steps of how transactions are executed in the transactional data processing flow:

  1. Initialization (initTransactions): The producer contacts the Transaction Coordinator to register its transactional.id. The Coordinator allocates a new Producer ID (PID) and raises the epoch value to block hanging old producers (zombie fencing).
  2. Begin Transaction (beginTransaction): The producer marks the start of a new transaction locally in client memory.
  3. Send Messages (send): The producer sends messages to the output topic. Before writing data to physical partitions, the producer tells the Coordinator to record those partitions in the __transaction_state journal. This step prevents brokers from writing data from unregistered transactions.
  4. Send Offsets (sendOffsetsToTransaction): Instead of sending commit offsets directly to the __consumer_offsets topic separately, the producer sends those input offsets to the Transaction Coordinator. The Coordinator registers them into the active transaction.
  5. Commit Transaction (commitTransaction): The producer asks the Coordinator to finalize the transaction. The Coordinator writes the PrepareCommit status to the __transaction_state journal.
  6. Writing Transaction Markers: The Coordinator writes special marker documents called Commit Markers or Abort Markers to all output topic partitions and offset topics participating in that transaction.
  7. Done: After all transaction markers are successfully written to all partitions, the Coordinator updates the transaction status in the __transaction_state journal to CompleteCommit. The transaction is fully complete.

State Diagram: Transaction Lifecycle #

The Kafka transaction lifecycle transitions through a series of states under the Transaction Coordinator’s supervision, as illustrated in the following state diagram:

stateDiagram-v2
    [*] --> Empty: initTransactions()
    Empty --> Ongoing: beginTransaction()
    Ongoing --> Ongoing: send() / sendOffsets()
    Ongoing --> PrepareCommit: commitTransaction()
    Ongoing --> PrepareAbort: abortTransaction()
    PrepareCommit --> CompleteCommit: Write Commit Marker to Partitions
    PrepareAbort --> CompleteAbort: Write Abort Marker to Partitions
    CompleteCommit --> Empty: Done
    CompleteAbort --> Empty: Done

Consumer-Level Configuration: isolation.level #

Transaction atomicity guarantees on the producer side are useless if our downstream consumer applications process every incoming message directly without checking the message’s transaction status. Therefore, the Exactly-Once guarantee requires us to align consumer configuration through the isolation.level property.

There are two configuration values for isolation.level:

1. read_uncommitted (Default) #

Consumers read all messages written in the broker’s partition log in physical offset order. This means consumers read messages from ongoing transactions as well as messages from aborted transactions. This scenario is very dangerous if our application isn’t tolerant of dirty reads.

2. read_committed #

Consumers only read:

  • Non-transactional messages (ordinary messages without transactions).
  • Transactional messages from transactions that have successfully received a Commit Marker.
  • Consumers actively skip transactional messages marked with an Abort Marker.

How Do Consumers Know Transaction Boundaries? #

The Kafka broker uses an internal marker called the Last Stable Offset (LSO). LSO is the smallest offset of an ongoing transaction. Consumers configured with isolation.level=read_committed aren’t allowed to read messages beyond the LSO boundary, even if new non-transactional messages are written after that ongoing transaction. This ensures consumers don’t get ahead of uncommitted transaction processes.

Java Implementation: The Transactional read-process-write Pattern #

Here’s a code comparison between the duplication-prone non-transactional data processing flow (anti-pattern) and the safe Exactly-Once Semantics processing flow:

// ANTI-PATTERN: Data processing flow without transactional guarantees
// If a crash happens after send() but before commitSync(), duplicate messages will be resent
public class VulnerableStreamProcessor {
    public void process(KafkaConsumer<String, String> consumer, KafkaProducer<String, String> producer) {
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                // 1. Process business data
                String outputValue = transform(record.value());
                
                // 2. Send to the output topic
                producer.send(new ProducerRecord<>("output-topic", record.key(), outputValue));
            }
            // ✗ Commit offsets separately. If a crash happens before this line executes,
            // the data in output-topic is already written, triggering data duplication on the next reload.
            consumer.commitSync(); 
        }
    }
    private String transform(String in) { return in.toUpperCase(); }
}

// CORRECT: Using Exactly-Once Semantics (EOS) with the Transaction API
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.TopicPartition;
import java.time.Duration;
import java.util.*;

public class SecureStreamProcessor {
    public void process() {
        // 1. Transactional Producer Configuration
        Properties prodProps = new Properties();
        prodProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        prodProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        prodProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        
        // ✓ CORRECT: transactional.id must be set to enable the Transaction API
        prodProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "processor-tx-id-0");
        prodProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true"); // Mandatory

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

        // 2. Read Committed Consumer Configuration
        Properties consProps = new Properties();
        consProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        consProps.put(ConsumerConfig.GROUP_ID_CONFIG, "processor-group");
        consProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        consProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        
        // ✓ CORRECT: Consumers must be set to only read successfully committed transactions
        consProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
        consProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); // Disable auto-commit

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consProps);
        consumer.subscribe(Collections.singletonList("input-topic"));

        // Initialize transactions at the coordinator
        producer.initTransactions();

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                if (records.isEmpty()) continue;

                // ✓ Start a new atomic transaction
                producer.beginTransaction();

                try {
                    Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();

                    for (ConsumerRecord<String, String> record : records) {
                        // Data transformation process
                        String outputValue = record.value().toUpperCase();

                        // Send data to the output topic within the transaction
                        producer.send(new ProducerRecord<>("output-topic", record.key(), outputValue));

                        // Record the next input message offset to read
                        offsetsToCommit.put(
                            new TopicPartition(record.topic(), record.partition()),
                            new OffsetAndMetadata(record.offset() + 1)
                        );
                    }

                    // ✓ Send the offset commit to the transaction coordinator to be included in the atomic transaction
                    producer.sendOffsetsToTransaction(offsetsToCommit, consumer.groupMetadata());

                    // ✓ Commit all operations: data sent and offsets committed simultaneously
                    producer.commitTransaction();

                } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
                    // Fatal unrecoverable errors, we must close the producer
                    producer.close();
                    break;
                } catch (KafkaException e) {
                    // Transient errors, abort the transaction and retry from the previous offset
                    System.err.println("Failed to process batch, rolling back transaction: " + e.getMessage());
                    producer.abortTransaction();
                }
            }
        } finally {
            consumer.close();
            producer.close();
        }
    }
}

Important Limitations and Warnings in Using EOS #

Although Exactly-Once Semantics provides an incredibly strong data safety guarantee, we must understand some of its architectural limitations to avoid system design mistakes in production:

1. Only Applies Within the Kafka Ecosystem (Kafka-Only Boundary) #

Kafka EOS only guarantees atomic operations that happen within the internal boundaries of the Kafka cluster.

  • Limitation: If in our read-process-write processing thread we write data to an external SQL database (e.g., MySQL) or call third-party REST APIs, the Kafka transaction cannot guarantee the atomicity of those external systems. If the Kafka transaction is aborted, data already written to the MySQL database won’t be automatically rolled back by Kafka. We still need to implement the Outbox Pattern or manual idempotency handling on the database side.

2. Transaction Latency Overhead #

Using transactions adds network and broker I/O overhead:

  • Every transaction requires a journal write to the __transaction_state topic.
  • Every destination partition must receive Commit/Abort Markers.
  • read_committed consumers experience a slight read delay because they must wait for transactions to complete (committed) until reaching the Last Stable Offset (LSO) boundary.
  • Recommendation: Don’t do transactions with overly small batch sizes (e.g., per 1 message). Do reasonable batch processing (e.g., per 100-1000 messages) to proportionally spread the transaction overhead across many messages.

Summary #

  • Exactly-Once Semantics: The exactly-once data processing guarantee combining the Idempotent Producer feature with the transactional API.
  • read-process-write: The data processing flow where input messages are read, processed, and output messages are written to the destination topic along with input offsets in one atomic transaction.
  • Transaction Coordinator: The designated broker specifically recording and monitoring transaction lifecycle status into the internal __transaction_state topic.
  • transactional.id: The unique, persistent producer identity guaranteeing protection against duplicate producer emergence (zombie fencing) after restarts.
  • read_committed: The mandatory consumer setting ensuring consumers only process transactional messages that have successfully received commit markers.
  • Kafka-Only Boundary: Kafka transactions don’t cover external storage systems (like SQL databases or external APIs) outside the Kafka broker scope.

← Previous: Idempotent Producer Next: Over-Partitioning →

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