Transaction API #

When we design real-time data processing systems, the biggest challenge appears when data must be atomically distributed to various different topics and partitions. Without transaction guarantees, system failures mid-data-delivery leave partial half-sent data, producing data inconsistencies that are very hard to fix manually. To solve this coordination challenge, Apache Kafka provides the Transaction API. This API allows producer applications to act as distributed transaction agents capable of sending a group of messages to various topics atomically (all messages permanently stored or none at all). Through this article, we’ll dissect the internal architecture behind the Transaction API, how the Transaction Coordinator component works, the transaction lifecycle flow, Zombie Fencing protection, and safe production Java implementation code.

Internal Architecture: How Are Kafka Transactions Coordinated? #

Behind the scenes, the Kafka transaction architecture is based on a variation of the classic distributed Two-Phase Commit (2PC) protocol, highly optimized for high performance. Instead of relying on slow external transaction managers (like XA transactions based on the X/Open protocol requiring third-party coordinators such as Atomikos or Bitronix), Kafka uses internal cluster components to manage transaction status independently. This fundamental difference is crucial: in XA transactions, if the external coordinator dies mid-way, all involved databases can be locked in a blocking locks pending state. In Kafka, because the coordinator is the broker itself and its replication is managed by partition logs, coordinator recovery runs very fast through standard partition leader failover mechanisms without locking data partitions.

This transaction architecture is supported by two main components:

1. Transaction Coordinator #

Similar to the Group Coordinator responsible for managing consumer group membership, the Transaction Coordinator is an internal module running inside the Kafka broker.

Coordinator Selection: Every transactional producer must declare a unique configuration string called transactional.id. Kafka selects which broker becomes the coordinator for that producer by hashing the transactional.id value, then mapping the result to a partition of the internal transaction log topic.
Coordinator Duties: Managing Producer ID (PID) allocation, recording the registration of topic partitions involved in active transactions, writing transaction status to the status log, and spreading Commit/Abort Markers to the destination partition leader brokers when the transaction ends.

2. The Internal `__transaction_state` Topic #

All transaction status transitions (like Ongoing, PrepareCommit, CompleteCommit) are permanently recorded to an internal log topic named __transaction_state.

This topic has many partitions (by default 50 partitions) and is configured with Log Compaction policy and a high Replication Factor (default 3 in production) to guarantee transaction status resilience from single broker failures. Important properties like transaction.state.log.replication.factor=3 and transaction.state.log.min.isr=2 ensure every transaction status change is safely duplicated before the coordinator proceeds to the next step.
Writes to this topic are append-only, ensuring transaction status write latency is very fast. The Log Compaction process trims old transaction logs already successfully committed or aborted, leaving only the most current active or committed transactional states to save broker disk storage.

sequenceDiagram
    autonumber
    participant Prod as Producer Client
    participant TC as Transaction Coordinator
    participant TopicLog as __transaction_state Topic
    participant PartLeader as Destination Partition Broker

    Note over Prod, TC: "Step 1: Transaction Initialization"
    Prod->>TC: InitProducerId (transactional.id="tx_prod_01")
    TC->>TopicLog: Record new PID & Epoch to the log
    TC-->>Prod: Return PID & Epoch (Fencing old zombies)

    Note over Prod, PartLeader: "Step 2: Transactional Delivery Process"
    Prod->>TC: AddPartitionsToTxnRequest (Registering destination partitions)
    TC->>TopicLog: Record the partition list with ONGOING status
    Prod->>PartLeader: ProduceRequest (Send Data to Partitions A & B)
    PartLeader->>PartLeader: Write Data to disk (Not yet stable for consumers)

    Note over Prod, PartLeader: "Step 3: Transaction Commit (2-Phase Commit)"
    Prod->>TC: EndTxnRequest (Commit)
    TC->>TopicLog: Record PREPARE_COMMIT status
    TC->>PartLeader: WriteTxnMarkerRequest (COMMIT MARKER)
    PartLeader->>PartLeader: Write Commit Marker to the Partition Log
    TC->>TopicLog: Record COMPLETE_COMMIT status (Transaction Successful!)
    TC-->>Prod: Return Commit Success

Transaction Lifecycle #

To use the Transaction API correctly, our producer must interact with the coordinator through a series of API calls with a strict logical order. Here’s the functional breakdown of each API:

1. `initTransactions()` #

This function must be called exactly once when the producer application starts, before any data is sent.

Duty: Contacts the Transaction Coordinator to register the transactional.id. The Coordinator allocates a new PID or finds the old PID bound to that transaction ID, increments the Producer Epoch value, and aborts all hanging transactions from unfinished old producer sessions.

2. `beginTransaction()` #

Called every time we want to start a new transaction batch.

Duty: Sets the producer client’s internal transaction status to active. The client sends no signal to the coordinator broker at this stage to save network bandwidth.

3. `send()` #

Sends messages to one or several destination topics.

Duty: Before the physical message is sent to the destination partition leader broker, the producer client library automatically sends an AddPartitionsToTxnRequest request to the Transaction Coordinator to register that partition into the active transaction list. After it’s registered in __transaction_state, the physical data is then sent to the destination broker.

4. `sendOffsetsToTransaction()` #

A special function used if our application acts as a stream processor (Read-Process-Write).

Duty: Sends consumer offset commits directly to the Transaction Coordinator to be stored in the internal __consumer_offsets topic as part of the producer transaction. This ensures old message reads and new message sends finish simultaneously in one single atomic transaction. Behind the scenes, the coordinator registers the __consumer_offsets topic partitions into the transaction list, similar to regular data partitions.

5. `commitTransaction()` #

Called when the entire data delivery process succeeds without obstacles.

Duty: Sends an EndTxnRequest request with COMMIT status to the coordinator. The coordinator writes the PrepareCommit status, spreads the Commit Marker to destination partition brokers, and finally closes the transaction by writing the CompleteCommit status.

6. `abortTransaction()` #

Called inside the catch block if an error or business logic failure happens.

Duty: Sends an EndTxnRequest request with ABORT status to the coordinator. The coordinator writes the PrepareAbort status, spreads the Abort Marker to destination partitions so transactional messages are silently discarded by read_committed consumers, then writes the CompleteAbort status.

Transaction Timeout Challenges #

Every Kafka transaction is limited by a certain duration to prevent transactions hanging forever from producers that crashed without calling abortTransaction(). This limit is set by the transaction.timeout.ms property (default 60000 ms or 1 minute). If this limit is exceeded while the transaction status is still Ongoing, the coordinator unilaterally aborts that transaction in the __transaction_state topic and spreads the Abort Marker. Producers trying to continue sending data on the same transaction receive TransactionTimeoutException.

Integration with Kafka Streams #

For developers using the Kafka Streams framework, this Transaction API is managed automatically under the hood. We just turn on the processing.guarantee="exactly_once_v2" configuration parameter. Kafka Streams dynamically creates transactional producer instances for each stream thread, manages unique transactional.ids based on task IDs, and periodically calls the transactional commit cycle beginTransaction() and commitTransaction() after each data processing batch is calculated.

The Zombie Fencing Concept: Protection from Split-Brain #

In distributed system architecture, one of the hardest failures to detect is the Split-Brain problem or the appearance of a Zombie Producer.

Scenario: Imagine a transactional producer Prod_A is running. Suddenly, the JVM server experiences a very long Garbage Collection pause (GC Pause) (for example, 1 minute).
Impact: The Transaction Coordinator considers Prod_A dead because it didn’t respond. The orchestration system (like Kubernetes) starts a new producer instance Prod_B with the same transactional.id to take over the task.
Problem: After Prod_B is active and starts a new transaction, the GC pause on Prod_A finishes. Prod_A (now a zombie) wakes up and tries to send its pending remaining transactional messages to the broker. If allowed, this damages data consistency.

Kafka overcomes this zombie danger using the Zombie Fencing mechanism:

sequenceDiagram
    autonumber
    participant Prod_A as "Prod_A (Epoch 1)"
    participant Prod_B as "Prod_B (Epoch 2)"
    participant Coord as "Coordinator"
    participant Broker as "Broker"
    
    Note over Prod_A: Experiences a GC Pause
    Coord->>Coord: Considers Prod_A dead
    Prod_B->>Coord: InitTransactions
    Coord->>Coord: Epoch incremented to 2
    Note over Prod_A: Wakes from GC
    Prod_A->>Broker: Sending data (Epoch 1)
    Broker-->>Prod_A: REJECTS DATA! (throws ProducerFencedException)

When Prod_B calls initTransactions(), the Transaction Coordinator increments the Epoch value (producer generation number) from 1 to 2 for that PID. When zombie Prod_A tries to send data using Epoch 1, the partition leader broker immediately rejects that delivery and returns the ProducerFencedException error. Prod_A realizes it has been replaced, then cleanly stops its operations.

Java SDK Implementation Code: Safe Atomic Transactions #

Here’s a complete implementation example of using the Transaction API with the Java SDK. This example demonstrates how to correctly configure transaction parameters, do initialization, and manage transaction failures using try-catch-finally blocks.

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.errors.ProducerFencedException;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

public class SecureTransactionApiExample {
    private static final Logger log = LoggerFactory.getLogger(SecureTransactionApiExample.class);
    private static final String BOOTSTRAP_SERVERS = "localhost:9092";
    private static final String TRANSACTIONAL_ID = "finance-payment-tx-01";
    private static final String TOPIC_A = "user-balances";
    private static final String TOPIC_B = "transaction-ledger";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // =====================================================================
        // TRANSACTION API SPECIFIC CONFIGURATION
        // =====================================================================
        // ✓ transactional.id must be set and unique per producer instance
        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, TRANSACTIONAL_ID);
        
        // ✓ Automatically enables producer idempotence (enable.idempotence=true)
        // and forces acks=all for maximum durability guarantees.
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
        
        // Transaction wait limit, automatically aborted by the broker if hanging
        props.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "60000"); // 1 Minute

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

        log.info("Step 1: Initializing the transaction and killing old zombie producers...");
        // initTransactions() contacts the coordinator, allocates a PID,
        // increments the epoch, and aborts leftover hanging transactions.
        producer.initTransactions();

        try {
            log.info("Step 2: Starting a new transaction block...");
            producer.beginTransaction();

            String userId = "usr_404";
            String debitPayload = "{\"userId\":\"" + userId + "\",\"action\":\"DEBIT\",\"amount\":250000}";
            String ledgerPayload = "{\"txId\":\"tx_999\",\"status\":\"SUCCESS\",\"userId\":\"" + userId + "\"}";

            log.info("Step 3: Sending transactional messages to Topic A...");
            // Partitions for TOPIC_A are automatically registered to the coordinator before data is sent
            producer.send(new ProducerRecord<>(TOPIC_A, userId, debitPayload));

            log.info("Step 4: Sending transactional messages to Topic B...");
            // Partitions for TOPIC_B are automatically registered to the coordinator before data is sent
            producer.send(new ProducerRecord<>(TOPIC_B, userId, ledgerPayload));

            // Simulate an additional business logic check before committing
            if (System.currentTimeMillis() % 2 == 0) {
                // For random failure demos
                throw new RuntimeException("Simulated business logic failure.");
            }

            log.info("Step 5: Committing the transaction atomically...");
            // Send the EndTxn request to the coordinator to write the Commit Marker
            producer.commitTransaction();
            log.info("✓ Transaction successfully declared atomically to both topics.");

        } catch (ProducerFencedException e) {
            // ZOMBIE FENCING EFFECT:
            // This exception happens if another producer instance with the same transactional.id
            // has become active and taken over transaction ownership.
            log.error("✗ Failed: This producer was detected as a Zombie! Forcefully stopping the application.", e);
            // Don't call abortTransaction() because the coordinator has rejected this producer.
            // The best step is shutting down the application for split-brain investigation.
            System.exit(1);
        } catch (Exception e) {
            log.error("✗ A failure occurred during the transaction. Doing rollback / abort...", e);
            try {
                // Send the EndTxn request to the coordinator to write the Abort Marker
                producer.abortTransaction();
                log.info("✓ Transaction rollback successfully completed on the broker.");
            } catch (Exception abortException) {
                log.error("✗ Failed to abort the transaction on the broker: ", abortException);
            }
        } finally {
            log.info("Cleanly closing the producer connection.");
            producer.close();
        }
    }
}

Summary #

Transaction API — The Kafka programming interface allowing producers to send a group of messages to various topic partitions atomically (All-or-Nothing).
Transaction Coordinator — The internal Kafka broker module responsible for coordinating client transaction status based on the hash function of the transactional.id configuration.
The __transaction_state Topic — The compressed internal log topic where the coordinator records every transaction status transition before emitting marker signals.
Zombie Fencing — The protection mechanism blocking leftover deliveries from old (zombie) producers by rejecting Epochs lower than the latest registered Epoch on the broker.
Two-Phase Commit (2PC) — Kafka’s internal coordination protocol dividing transaction commits into a PrepareCommit record writing phase followed by Commit Markers spreading.
Commit/Abort Markers — Special control messages the coordinator writes to physical partition logs as transaction end boundary markers for downstream consumers.
initTransactions() — The mandatory initialization function for registering the transaction ID, incrementing the producer Epoch, and cleaning up hanging transactions from previous sessions.

← Previous: Exactly-Once Next: Producer Consumer Transaction →

Transaction API #

Internal Architecture: How Are Kafka Transactions Coordinated? #

1. Transaction Coordinator #

2. The Internal __transaction_state Topic #

Transaction Lifecycle #

1. initTransactions() #

2. beginTransaction() #

3. send() #

4. sendOffsetsToTransaction() #

5. commitTransaction() #

6. abortTransaction() #