Learn Apache Kafka - Kafka Transactions

#Introduction

In episode 8 you already understood delivery semantics and where exactly-once sits in the at-most-once and at-least-once spectrum. Now it's time to practice the most technical part: Kafka transactions — the mechanism that makes sending records to multiple partitions atomic.

Kafka transactions solve a problem that the idempotent producer can't handle on its own. The idempotent producer prevents duplicates within a single partition, but it doesn't make multiple writes to different partitions or topics all-or-nothing. Transactions close that gap by introducing a transactional id and a transaction coordinator.

Episode 9 walks you through the five core methods of the transaction API, the read-process-write pattern that is the backbone of stream processing, the configuration that governs transaction behavior, and the performance costs you must weigh before deciding to use transactions.

#Transaction API

#Lifecycle of Five Methods

Kafka transactions are driven through five methods on KafkaProducer that follow a strict sequence:

producer.initTransactions();
producer.beginTransaction();
producer.send(new ProducerRecord<>("orders", "order-001"));
producer.send(new ProducerRecord<>("order-events", "order-001"));
producer.commitTransaction();

• initTransactions(): initializes the transactional producer and registers the transactional.id with the coordinator.
• beginTransaction(): marks the start of a transaction; all sends after this become part of the transaction.
• send(): sends records as usual, but they aren't visible to read_committed consumers until commit.
• commitTransaction(): flushes all records and ends the transaction successfully.
• abortTransaction(): cancels the transaction; all sent records are discarded and never become visible.

Every transaction must end with a commit or abort. Leaving a transaction hanging makes the coordinator time out and abort it automatically.

#Handling Failures

Never let an exception leave a transaction in an unclear state:

producer.initTransactions();
try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("orders", "order-001", "created"));
    producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException e) {
    producer.close();
} catch (KafkaException e) {
    producer.abortTransaction();
}

If a transaction is fenced (another producer uses the same transactional.id), the old instance must be closed. For other errors, abortTransaction() cleans up the state so the coordinator doesn't hold resources.

#Transactional Patterns

#Read-Process-Write

The most common pattern: a consumer reads records, the application processes them, then a producer writes the results — all within one transaction. The processing results and consumed offsets are sent together, so rolling back one cancels both:

consumer.subscribe(List.of("input-topic"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    producer.beginTransaction();
    for (ConsumerRecord<String, String> record : records) {
        producer.send(new ProducerRecord<>("output-topic", record.key(), "processed:" + record.value()));
        producer.sendOffsetsToTransaction(currentOffsets(records), consumer.groupMetadata());
    }
    producer.commitTransaction();
}

Notice sendOffsetsToTransaction: offsets are sent as part of the transaction, so re-consumption doesn't happen when processing fails — this is the basis of end-to-end exactly-once in Kafka Streams.

#Atomic Multi-Partition Writes

A single transaction can write to any partition and topic. For example, an order service writes to orders and inventory-updates at the same time; both become visible to read_committed consumers only when both are successfully committed. A failure in either aborts everything.

#Exactly-Once Stream Processing

In stream processing, transactions combine internal state writes (changelog) with output writes. Kafka Streams uses this pattern automatically with processing.guarantee=exactly_once_v2 — you don't need to write manual transaction code; the library handles it.

#Transaction Configuration

#transactional.id

transactional.id must be unique and stable per application instance. A stable value lets the coordinator detect dead old producers and reject their writes (fencing):

transactional.id=order-processor-1
transaction.timeout.ms=60000
enable.idempotence=true
acks=all

If two instances use the same transactional.id at the same time, the second instance will fence off the first — preventing two processes from writing interleaved. For that reason make sure transactional.id is unique per instance, for example order-processor-1 for the first node and order-processor-2 for the second.

#isolation.level on the Consumer

The consumer chooses whether unfinished transactions may be read:

• read_uncommitted (default): sees all records, including those still in a transaction or already aborted.
• read_committed: only sees records from committed transactions; records from aborted transactions are discarded.

End-to-end consistency is only achieved when producers use transactions and consumers use isolation.level=read_committed together.

#Performance Considerations

#The Cost of Transactions

Transactions are not free. Each transaction involves at least one extra round-trip to the transaction coordinator, and a commit requires writing markers to the internal __transaction_state topic. The impact:

• Throughput: lower than a non-transactional producer, especially for small, frequent transactions.
• Latency: higher because it waits for transaction status synchronization across partitions.
• Resources: the coordinator holds memory and disk for in-flight transaction state.

#Wise Trade-Offs

Don't wrap every record in its own transaction. Best practices:

• Batch many records into one transaction — for example one transaction per consumption batch, not per record.
• Only use transactions for flows that genuinely need atomicity: read-process-write, critical state synchronization, or cross-topic synchronization.
• For simple pipelines tolerant of duplicates, at-least-once remains far cheaper.
• Monitor transaction.count and coordinator metrics to detect hanging transactions.

#Closing

In this episode 9 you've run the five transaction API methods, understood the read-process-write pattern with sendOffsetsToTransaction, configured transactional.id and isolation.level, and weighed the throughput and latency costs of transactions.

The key takeaways:

• Transactions make cross-partition writes atomic: all visible or none at all.
• The lifecycle is always initTransactions, beginTransaction, send, then commit or abort.
• sendOffsetsToTransaction binds consumed offsets to the transaction for end-to-end EOS.
• transactional.id must be unique and stable; fencing prevents two instances from writing at once.
• Consumers must use isolation.level=read_committed for full consistency.
• Batch many records per transaction; don't use one transaction per record.

In the next episode 10 we'll discuss log compaction — the opposite of retention delete. You'll learn how Kafka retains the latest value per key with tombstones, the min.compaction.lag.ms, max.compaction.lag.ms, and min.cleanable.dirty.ratio parameters, and the changelog topic, materialized views, and state store use cases. Transactions and compaction are the two foundations of Kafka's state storage that you'll keep running into.