Exactly-Once Semantics

Architecture

A single transaction touches three brokers (or roles): the producer, the transaction coordinator, and the data partition leaders.

Key Numbers

Idempotent Producer: PID + Epoch + Sequence

With enable.idempotence=true, every batch carries a triple: producer ID (assigned by broker on InitProducerId), epoch, and a per-partition monotonic sequence number. The broker tracks the last 5 sequences it accepted per (PID, partition) and dedupes retries.

// Producer config — idempotent
Properties props = new Properties();
props.put("bootstrap.servers", "kafka-1:9092,kafka-2:9092");
props.put("enable.idempotence", "true");
// these are forced by idempotence, but be explicit:
props.put("acks", "all");
props.put("max.in.flight.requests.per.connection", "5");
props.put("retries", Integer.MAX_VALUE);

KafkaProducer<String,String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("orders", key, value));

What the broker sees on retry: the same (PID=42, epoch=0, seq=17) batch arrives twice. The first write succeeds and the broker remembers seq=17 in the leader's producer state. The second arrives with a duplicate sequence; the broker returns the original offset without writing again. Out-of-order sequences (seq=20 when last was 17) are rejected with OutOfOrderSequenceException — the producer must abort and reset.

Transactions API

A transaction wraps multiple sends across multiple partitions, plus the consumer offset commit, into one atomic unit.

props.put("transactional.id", "orders-processor-1");
props.put("isolation.level", "read_committed"); // for the consumer side
KafkaProducer<String,String> producer = new KafkaProducer<>(props);
producer.initTransactions();  // fences any prior producer with same transactional.id

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

  producer.beginTransaction();
  try {
    Map<TopicPartition,OffsetAndMetadata> offsets = new HashMap<>();
    for (ConsumerRecord<String,String> r : records) {
      String out = transform(r.value());
      producer.send(new ProducerRecord<>("orders-enriched", r.key(), out));
      offsets.put(new TopicPartition(r.topic(), r.partition()),
                  new OffsetAndMetadata(r.offset() + 1));
    }
    producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
    producer.commitTransaction();
  } catch (ProducerFencedException e) {
    // a newer instance took over with the same transactional.id — exit
    producer.close();
    return;
  } catch (KafkaException e) {
    producer.abortTransaction();
  }
}

initTransactions() does two things: it discovers the transaction coordinator (the broker hosting the leader of the __transaction_state partition for this transactional.id, picked by hashing the id), and it bumps the epoch to fence any zombie. sendOffsetsToTransaction() is the magic ingredient — it writes to __consumer_offsets as a transactional record, so the offset commit is visible to other consumers only when the transaction commits.

The __transaction_state Topic

An internal compacted topic, one record per (transactional.id, state). The Transaction Coordinator is the broker leading the partition for a given id.

# Inspect transaction state (DEBUG only — internal topic)
$ kafka-console-consumer.sh --bootstrap-server kafka-1:9092 \
    --topic __transaction_state --from-beginning \
    --formatter "kafka.coordinator.transaction.TransactionLog\$TransactionLogMessageFormatter"

orders-processor-1::TransactionMetadata(
  transactionalId=orders-processor-1,
  producerId=1003,
  producerEpoch=4,
  txnTimeoutMs=60000,
  state=Ongoing,
  topicPartitions=Set(orders-enriched-0, orders-enriched-1, __consumer_offsets-12),
  txnStartTimestamp=1714742103000,
  txnLastUpdateTimestamp=1714742104100
)

State transitions: Empty → Ongoing → PrepareCommit → CompleteCommit (or PrepareAbort → CompleteAbort). Once PrepareCommit is durable on the coordinator's log, the transaction is decided — even if every involved broker dies and recovers, replay of the transaction log will resume and write commit markers.

Two-Phase Commit Protocol

A simplified trace of one successful transaction with sends to two partitions and an offset commit.

1. Producer  -> Coordinator  : InitProducerId(transactional.id="orders-processor-1")
   Coordinator -> Producer    : PID=1003, epoch=4

2. Producer  -> Coordinator  : AddPartitionsToTxn([orders-enriched-0, orders-enriched-1])
   Coordinator writes "Ongoing + partitions" to __transaction_state

3. Producer  -> Leader(orders-enriched-0) : ProduceRequest (transactional=true, PID=1003, epoch=4, seq=0..N)
   Producer  -> Leader(orders-enriched-1) : ProduceRequest ...

4. Producer  -> Coordinator  : AddOffsetsToTxn(__consumer_offsets-12, group=orders-group)
   Producer  -> Coordinator  : TxnOffsetCommit(offsets)

5. Producer  -> Coordinator  : EndTxn(commit=true)
   Coordinator writes "PrepareCommit" to __transaction_state    <-- decision point
   Coordinator -> Leader(orders-enriched-0) : WriteTxnMarker(COMMIT, PID, epoch)
   Coordinator -> Leader(orders-enriched-1) : WriteTxnMarker(COMMIT, PID, epoch)
   Coordinator -> Leader(__consumer_offsets-12) : WriteTxnMarker(COMMIT, PID, epoch)
   Coordinator writes "CompleteCommit" to __transaction_state

Consumer Side: isolation.level

The producer's transactional output is invisible to consumers unless they opt in.

props.put("isolation.level", "read_committed");
// default is read_uncommitted, which sees aborted records too

KafkaConsumer<String,String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("orders-enriched"));

With read_committed, the consumer's fetcher tracks the Last Stable Offset (LSO) per partition. The LSO is the offset before the earliest still-open transaction. Records past the LSO are buffered until their transaction commits or aborts; aborted records are filtered out via the abort index in the log segment. This adds latency proportional to the longest open transaction.

Tradeoffs and When Not to Use

Frequently Asked Questions