Hands-on Practical: Implementing Transactional Writes

Distributed systems inherently face the risk of partial failures where a process crashes after writing to one partition but before writing to another. This partial state is unacceptable in scenarios requiring strict data integrity, such as financial ledger updates or inventory management. Apache Kafka addresses this through its Transaction API, which enables atomic writes across multiple partitions. This capability ensures that a group of messages is either fully visible to downstream consumers or completely effectively discarded.

The Transaction Protocol

Kafka transactions rely on a two-phase commit protocol adapted for streaming. This protocol introduces a specialized component called the Transaction Coordinator. The coordinator maintains the state of active transactions in an internal topic named __transaction_state.

When a producer initiates a transaction, it registers with the coordinator. The coordinator assigns a producer ID (PID) and an epoch. During the write phase, the producer sends messages to user topics as usual, but these messages are not immediately considered "committed." Once the producer issues a commit command, the coordinator writes a "Prepare Commit" message to the transaction log. Subsequently, the coordinator writes "Commit Markers" to all partitions involved in the transaction.

The following diagram illustrates the interaction between the Producer, the Transaction Coordinator, and the Topic Partitions during a successful commit sequence.

The sequence of operations ensuring atomic writes across distinct partitions via the Transaction Coordinator.

Configuring the Transactional Producer

To enable transactions, you must configure the transactional.id property in the producer. This ID must be static and unique for each producer instance. It allows the transaction coordinator to identify a specific producer across restarts. If a producer crashes and restarts with the same transactional.id, the coordinator fences off the old instance using an epoch mechanism, preventing "zombie" processes from corrupting data.

The enablement of transactions automatically enforces idempotence (enable.idempotence=true), ensuring that retries do not result in duplicates within a single partition.

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// The transactional.id is strictly required
props.put("transactional.id", "order-processor-01");

// High durability settings are recommended for transactional workloads
props.put("enable.idempotence", "true");
props.put("acks", "all");

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

Executing the Transaction Loop

The implementation of a transactional workflow follows a specific lifecycle: initialization, beginning the transaction, sending records, and finally committing or aborting. The initTransactions() method is called exactly once to register the producer with the coordinator and synchronize epochs.

This logic must be wrapped in strong error handling. Specifically, you must handle ProducerFencedException, which indicates that another instance with the same transactional.id has started, and AuthorizationException, which indicates permission failures.

// 1. Initialize the transaction context
producer.initTransactions();

try {
    // 2. Begin a new transaction
    producer.beginTransaction();

    // 3. Send records to multiple partitions
    ProducerRecord<String, String> recordA = 
        new ProducerRecord<>("financial-ledger", "Account-123", "Debit: $100");
    ProducerRecord<String, String> recordB = 
        new ProducerRecord<>("audit-log", "Account-123", "TransactionID: 998877");

    producer.send(recordA);
    producer.send(recordB);

    // 4. Commit the transaction
    producer.commitTransaction();

} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
    // Fatal errors: We cannot recover the producer
    producer.close();
    throw e;
} catch (KafkaException e) {
    // Non-fatal errors: Abort the transaction and retry if necessary
    producer.abortTransaction();
}

Consumer Isolation Levels

Writing transactionally is only half the equation. If consumers read data using the default configuration, they may see messages that are part of an open transaction or messages from an aborted transaction. To guarantee end-to-end consistency, consumers must be configured with the correct isolation level.

Setting isolation.level to read_committed ensures the consumer only reads messages that have been successfully committed. The consumer buffers messages internally until it encounters a control message (the commit marker). If it encounters an abort marker, the buffered messages are discarded.

The relationship between the producer's write status and the consumer's visibility is non-trivial. The following chart details the states a message batch can occupy.

Comparison of message availability for consumers based on the isolation level configuration.

Performance Implications

Transactional writes introduce overhead. This latency arises from the interaction with the Transaction Coordinator and the writing of control markers to the logs. The overhead is largely constant per transaction, regardless of the number of messages within that transaction. Therefore, the frequency of commits significantly impacts throughput.

We can define the effective throughput $T_{eff}$ relative to the raw throughput $T_{raw}$ and the transaction overhead $O_{tx}$ as:

$T_{eff} \approx \frac{N_{msgs} \times S_{msg}}{ (N_{msgs} \times T_{write}) + O_{tx} }$

Where $N_{msgs}$ is the number of messages per transaction. If $N_{msgs}$ is small (e.g., 1 message per transaction), $O_{tx}$ dominates, and performance degrades severely. Increasing the batch size within a beginTransaction / commitTransaction block amortizes the coordination cost.

However, increasing transaction size also increases the latency for read_committed consumers, as they cannot process data until the commit marker is received. Optimizing this requires balancing the throughput efficiency of large transactions against the latency requirements of downstream consumers.

Zombie Fencing and Epochs

A significant risk in distributed writes is the "zombie producer" scenario. This occurs when a producer pauses due to a network partition or long garbage collection (GC) pause. The system may assume the producer is dead and start a replacement. If the original producer wakes up and attempts to write, data corruption could occur.

Kafka handles this via epochs. When initTransactions() is called, the coordinator increments the epoch for that transactional.id.

$E_{current} = E_{previous} + 1$

Any write requests sent to the brokers include this epoch. If a broker receives a request with an epoch lower than the current one stored in the transaction metadata, it rejects the request with a ProducerFencedException. This mechanism guarantees that at any given time, only one valid instance of a producer with a specific ID can write to the log.