Definition
- Idempotent producer -- uses a broker-assigned producer id (PID) and per-partition sequence numbers to drop retry duplicates.
- Transaction coordinator -- a broker-side component that manages transactional state in an internal transaction log topic (
__transaction_state). - transactional.id -- a stable identifier that survives producer restarts; lets the coordinator track and fence a producer across sessions.
- Commit markers -- special control records written to every partition the transaction touched; they tell consumers which records belong to committed transactions.
- Epoch fencing -- when a new producer calls
initTransactions(), the coordinator bumps the epoch for that transactional.id, rejecting any writes from the old (zombie) instance. - read_committed -- consumer isolation level that delivers only committed transaction records and skips aborted or in-flight ones.
Imagine a National Bank of India clerk who moves money and records it in the ledger. Exactly-once is like a notarized atomic operation: either both the money movement and the ledger entry happen, or neither does. If the clerk is replaced mid-task, the old clerk's badge is revoked so he cannot sneak back and do the work twice. A special stamp on the ledger page says this block of entries is officially committed, and readers ignore entries without that stamp.
Understanding the Concepts
What are the three pieces of exactly-once?
Exactly-once semantics means each input record affects the output and the offsets exactly once, with no duplicates and no loss. It is built from three pieces.
package com.bank.eos;
public class ExactlyOncePieces {
public static void main(String[] args) {
System.out.println("Piece 1: Idempotent producer");
System.out.println(" PID + sequence -> broker drops retry duplicates");
System.out.println();
System.out.println("Piece 2: Kafka transactions");
System.out.println(" Atomic commit of output + offsets");
System.out.println();
System.out.println("Piece 3: Epoch fencing");
System.out.println(" Zombie producers rejected via epoch bump");
}
}
Piece 1: Idempotent producer
PID + sequence -> broker drops retry duplicates
Piece 2: Kafka transactions
Atomic commit of output + offsets
Piece 3: Epoch fencing
Zombie producers rejected via epoch bump
How does the idempotent producer work?
When idempotence is enabled, the broker assigns the producer a unique PID. The producer tags each record with a per-partition sequence number. If a retry sends the same sequence, the broker drops the duplicate.
package com.bank.producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import java.util.Properties;
public class IdempotentSetup {
public static Properties config() {
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092");
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
return props;
}
public static void main(String[] args) {
System.out.println("Idempotent producer:");
System.out.println(" Broker assigns PID=100");
System.out.println(" Record sent: PID=100, seq=5 -> written");
System.out.println(" ACK lost, retry: PID=100, seq=5 -> DUPLICATE, ack only");
}
}
Idempotent producer:
Broker assigns PID=100
Record sent: PID=100, seq=5 -> written
ACK lost, retry: PID=100, seq=5 -> DUPLICATE, ack only
How do transactions work?
Transactions extend idempotence across multiple partitions and tie the produce and the offset commit together.
package com.bank.eos;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
public class ExactlyOnceTransactionLoop {
public static void run(KafkaConsumer<String, String> consumer,
KafkaProducer<String, String> producer) {
producer.initTransactions();
while (true) {
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofMillis(100));
if (records.isEmpty()) continue;
producer.beginTransaction();
try {
records.forEach(record -> {
// Process: debit + credit
producer.send(new ProducerRecord<>("settlements",
record.key(), "{\"settled\":true}"));
});
// Commit offsets INSIDE the transaction
Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
records.partitions().forEach(tp -> {
long last = records.records(tp)
.get(records.records(tp).size() - 1).offset();
offsets.put(tp, new OffsetAndMetadata(last + 1));
});
producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
producer.commitTransaction();
} catch (Exception e) {
producer.abortTransaction();
}
}
}
}
How does the transaction coordinator work?
The transaction coordinator is a broker-side component that manages each transactional.id. It keeps state in the internal __transaction_state topic. On commit, it writes commit markers to every partition the transaction touched.
package com.bank.eos;
public class TransactionCoordinatorSteps {
public static void main(String[] args) {
System.out.println("Transaction coordinator flow:");
System.out.println(" 1. Producer calls initTransactions()");
System.out.println(" Coordinator assigns/bumps epoch for tx.id");
System.out.println();
System.out.println(" 2. Producer calls beginTransaction()");
System.out.println(" Coordinator logs: tx OPEN");
System.out.println();
System.out.println(" 3. Producer sends records to partitions");
System.out.println(" Coordinator tracks which partitions are in tx");
System.out.println();
System.out.println(" 4. Producer calls commitTransaction()");
System.out.println(" Coordinator logs: tx PREPARE_COMMIT");
System.out.println(" Coordinator writes COMMIT markers to all partitions");
System.out.println(" Coordinator logs: tx COMMITTED");
}
}
Transaction coordinator flow:
1. Producer calls initTransactions()
Coordinator assigns/bumps epoch for tx.id
2. Producer calls beginTransaction()
Coordinator logs: tx OPEN
3. Producer sends records to partitions
Coordinator tracks which partitions are in tx
4. Producer calls commitTransaction()
Coordinator logs: tx PREPARE_COMMIT
Coordinator writes COMMIT markers to all partitions
Coordinator logs: tx COMMITTED
What are commit markers?
Commit markers are special control records written to every partition the transaction touched, including the consumer offsets topic. They tell consumers that the records before them in this transaction are now committed.
package com.bank.eos;
public class CommitMarkers {
public static void main(String[] args) {
System.out.println("Commit markers written to:");
System.out.println(" settlements-0 : COMMIT marker");
System.out.println(" settlements-3 : COMMIT marker");
System.out.println(" __consumer_offsets-12 : COMMIT marker");
System.out.println();
System.out.println("Consumer with read_committed:");
System.out.println(" Sees records only after COMMIT marker");
System.out.println(" Skips records from ABORTED transactions");
}
}
Commit markers written to:
settlements-0 : COMMIT marker
settlements-3 : COMMIT marker
__consumer_offsets-12 : COMMIT marker
Consumer with read_committed:
Sees records only after COMMIT marker
Skips records from ABORTED transactions
How does epoch fencing stop zombies?
When a producer calls initTransactions(), the coordinator bumps the epoch for that transactional.id. If an old instance tries to write or commit with the old epoch, it is rejected with a ProducerFencedException.
package com.bank.eos;
public class EpochFencing {
public static void main(String[] args) {
System.out.println("Epoch fencing:");
System.out.println(" Instance A starts: tx.id=settlement-tx-1, epoch=1");
System.out.println(" Instance A hangs (zombie)");
System.out.println();
System.out.println(" Instance B starts: tx.id=settlement-tx-1, epoch=2");
System.out.println(" Coordinator bumps epoch from 1 to 2");
System.out.println();
System.out.println(" Instance A wakes up, tries to commit with epoch=1");
System.out.println(" Coordinator: REJECTED -- ProducerFencedException");
System.out.println(" Only Instance B (epoch=2) can write");
}
}
Epoch fencing:
Instance A starts: tx.id=settlement-tx-1, epoch=1
Instance A hangs (zombie)
Instance B starts: tx.id=settlement-tx-1, epoch=2
Coordinator bumps epoch from 1 to 2
Instance A wakes up, tries to commit with epoch=1
Coordinator: REJECTED -- ProducerFencedException
Only Instance B (epoch=2) can write
How does the consumer participate?
Consumers must set isolation.level=read_committed to see only committed transaction records.
package com.bank.consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import java.util.Properties;
public class ReadCommittedConfig {
public static Properties config() {
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "ledger-service");
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringDeserializer");
return props;
}
public static void main(String[] args) {
System.out.println("read_committed: sees only committed records");
System.out.println("read_uncommitted: sees everything (including aborted)");
}
}
read_committed: sees only committed records
read_uncommitted: sees everything (including aborted)
What are the limitations?
Exactly-once is end-to-end only within Kafka. The moment you write to an external system (database), you need that system to be idempotent or use an outbox pattern. Transactions add latency and coordinator overhead.
package com.bank.limitation;
public class ExactlyOnceLimitations {
public static void main(String[] args) {
System.out.println("Limitations:");
System.out.println(" 1. Only within Kafka -- cannot roll back PostgreSQL");
System.out.println(" 2. Transactions add latency (coordinator overhead)");
System.out.println(" 3. Use only where correctness demands it (money movement)");
System.out.println();
System.out.println("For external systems:");
System.out.println(" - Make DB writes idempotent (upsert on trade id)");
System.out.println(" - Or use the outbox pattern");
}
}
Limitations:
1. Only within Kafka -- cannot roll back PostgreSQL
2. Transactions add latency (coordinator overhead)
3. Use only where correctness demands it (money movement)
For external systems:
- Make DB writes idempotent (upsert on trade id)
- Or use the outbox pattern
What should a senior engineer know?
The consume-process-produce loop with sendOffsetsToTransaction is the foundation of exactly-once stream processing. The transaction coordinator uses a two-phase commit internally. Transactions add real cost, so reserve them for hard atomicity requirements. External side effects are outside the transaction boundary.
Visual Overview
Exactly-Once Processing End-to-End
Consumer Producer (tx.id) Coordinator
┌────────┐ ┌────────────────┐ ┌──────────────┐
│ poll() │ │ initTransactions│────────▶│ Assign epoch │
│ │ │ begin() │ │ epoch=2 │
│ records│──process──────▶│ send(settle-0) │ │ │
│ │ │ send(settle-3) │ │ Track parts │
│ offsets│──────────────▶│ sendOffsets() │ │ │
│ │ │ commit() │────────▶│ PREPARE │
└────────┘ └────────────────┘ │ Write COMMIT │
│ markers to: │
│ settle-0 │
│ settle-3 │
│ __offsets │
│ COMMITTED │
└──────────────┘
Zombie (epoch=1):
┌──────────────────┐ ┌──────────────┐
│ Old producer │──send──▶│ Coordinator │
│ epoch=1 │ │ epoch=1 < 2 │
│ │◀─REJECT─│ FENCED! │
└──────────────────┘ └──────────────┘
Key Takeaway
Exactly-once processing in Kafka is built from three pieces. The idempotent producer uses a PID and sequence numbers so the broker drops retry duplicates. Transactions wrap output records and consumed offsets in one atomic commit managed by the transaction coordinator, which writes commit markers to every partition the transaction touched. Epoch fencing ensures a zombie producer cannot sneak back and commit stale work.
On the reading side, consumers set isolation level to read_committed so they only see committed records and skip anything aborted or in-flight. The consume-process-produce loop with sendOffsetsToTransaction is the foundation of exactly-once stream processing.
At National Bank of India, this guaranteed that a transfer was never settled twice even under retries, crashes, or a zombie instance. The important caveat is that exactly-once is end-to-end only inside Kafka. For external systems like the ledger database, Rajesh makes writes idempotent or uses the outbox pattern.
Remember one sentence: exactly-once is idempotent writes plus a transaction that atomically commits output and offsets together, with epoch fencing stopping zombies and commit markers telling consumers which records are real.
---