---

🟢 Junior Level

30-Second Summary

Configuring Exactly-Once Semantics (EOS) in Apache Kafka requires coordinated alignment across all three tiers of the system—Producer, Brokers, and Consumer—combined with the programmatic Kafka Transaction API:

1. Producer Configuration:
   • Define a unique, persistent transactional.id.
   • Enable enable.idempotence = true (default in Kafka 3.0+).
   • Enforce durability: acks = all and retries = Integer.MAX_VALUE.
2. Broker Configuration:
   • Ensure transaction log fault tolerance: transaction.state.log.replication.factor = 3 and transaction.state.log.min.isr = 2.
3. Consumer Configuration:
   • Set the transactional isolation level: isolation.level = read_committed.
   • Disable automatic commit: enable.auto.commit = false.
4. Application Code (Read-Process-Write Pattern):
   • Execute consumption, business transformation, writing to output topics, and consumer offset commits inside a single atomic transaction via producer.sendOffsetsToTransaction().

---

Real-World Analogy

Think of Exactly-Once Semantics as an interbank accounting ledger transfer:

• You transfer $500 between accounts: funds must be debited from Checking Account A and credited to Savings Account B.
• If Account A is debited but the power fails before Account B is credited, the money vanishes into thin air.
• If Account B is credited but the server crashes before debiting Account A, duplicate money is created out of nothing.
• An atomic transaction ensures both entries appear in the master ledger simultaneously or both are completely rolled back, returning the system to its initial consistent state.

---

Read-Process-Write Topology

[ Input Topic A ] ──► [ Consumer.poll() ]
                            │
                            ▼
                    [ Business Logic ]
                            │
                            ▼
  ┌────────────────────────────────────────────────────────┐
  │              SINGLE ATOMIC KAFKA TRANSACTION           │
  │                                                        │
  │  1. [ Producer.send() ] ──────────► [ Output Topic B ] │
  │  2. [ sendOffsetsToTransaction() ] ► [ __consumer_offs ]
  └────────────────────────────────────────────────────────┘

---

🟡 Middle Level

Pure Java Client Implementation (Read-Process-Write)

In native Java applications, consumer offsets are committed directly through the producer’s transactional context via producer.sendOffsetsToTransaction():

Properties producerProps = new Properties();
producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
producerProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-stream-worker-1");
producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
producerProps.put(ProducerConfig.ACKS_CONFIG, "all");

Properties consumerProps = new Properties();
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processing-group");
consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);

// 1. One-time transaction initialization (executed once during application bootstrap)
producer.initTransactions();
consumer.subscribe(List.of("raw-orders"));

while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(200));
    if (records.isEmpty()) continue;

    try {
        // 2. Open atomic transaction boundary
        producer.beginTransaction();

        Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();

        for (ConsumerRecord<String, String> record : records) {
            String enrichedOrder = processOrder(record.value());
            producer.send(new ProducerRecord<>("validated-orders", record.key(), enrichedOrder));

            // Record next unconsumed offset
            offsetsToCommit.put(
                new TopicPartition(record.topic(), record.partition()),
                new OffsetAndMetadata(record.offset() + 1)
            );
        }

        // 3. Atomically bundle consumer offsets into this transaction (KIP-447)
        producer.sendOffsetsToTransaction(offsetsToCommit, consumer.groupMetadata());

        // 4. Two-Phase Commit: commits data records and consumer offsets atomically
        producer.commitTransaction();

    } catch (ProducerFencedException | InvalidProducerEpochException e) {
        log.error("Producer fenced out by newer instance (Zombie). Shutting down.", e);
        producer.close();
        break;
    } catch (KafkaException e) {
        log.warn("Transaction failed due to transient error. Aborting...", e);
        // 5. Rollback on failure: coordinator writes ABORT marker to partition logs
        producer.abortTransaction();
        resetToLastCommitted(consumer);
    }
}

---

Spring Boot 3 / Spring Kafka Configuration (KafkaTransactionManager)

In Spring Boot, EOS is configured declaratively by exposing a KafkaTransactionManager and annotating listener methods with @Transactional:

@Configuration
public class KafkaTxConfiguration {

    @Bean
    public DefaultKafkaProducerFactory<String, Object> producerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

        DefaultKafkaProducerFactory<String, Object> factory = new DefaultKafkaProducerFactory<>(props);
        // Defining a transactionIdPrefix automatically enables Kafka transactions
        factory.setTransactionIdPrefix("tx-order-");
        return factory;
    }

    @Bean
    public KafkaTransactionManager<String, Object> kafkaTransactionManager(
            ProducerFactory<String, Object> producerFactory) {
        return new KafkaTransactionManager<>(producerFactory);
    }
}

@Service
@RequiredArgsConstructor
public class TransactionalOrderConsumer {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    @Transactional("kafkaTransactionManager")
    @KafkaListener(topics = "raw-orders", groupId = "spring-tx-group",
                   containerFactory = "kafkaListenerContainerFactory")
    public void consume(ConsumerRecord<String, OrderDto> record) {
        OrderDto order = record.value();
        ProcessedOrder result = enrich(order);

        // Sent inside the active Kafka transaction
        kafkaTemplate.send("validated-orders", record.key(), result);
        // Consumer offset is committed atomically to __consumer_offsets when the method exits
    }
}

---

🔴 Senior Level

Zombie Fencing via Producer Epoch

In distributed container clusters, a producer pod might freeze during an extended Stop-the-World Garbage Collection pause. Kubernetes assumes the pod is dead and spawns a replacement pod on a different node. If the old pod unfreezes, two producer instances would attempt to publish records with the same identifier (Split-Brain hazard).

How Kafka Prevents Zombie Producers:

1. Upon startup, the new producer instance with the same transactional.id calls initTransactions().
2. The broker Transaction Coordinator increments the internal epoch counter: Producer Epoch = Epoch + 1.
3. The coordinator forcibly aborts any unresolved transactions left open by the previous epoch.
4. When the stalled zombie pod revives and attempts to invoke send() or commitTransaction() with its stale epoch, the broker rejects the request with a fatal ProducerFencedException.
5. The zombie process is forced to terminate immediately without corrupting log segments.

---

Evolution of EOS: KIP-447 (EOS v2 in Kafka 2.5+)

• EOS v1 (pre-Kafka 2.5): Required dedicated producer instances tied statically to specific consumer partitions. If a rebalance shifted partition assignments, active transactions hung or required restarting the entire application context.
• EOS v2 (KIP-447): Introduced producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata()). The method transmits the consumer group’s generationId, memberId, and groupInstanceId. The Transaction Coordinator verifies group membership directly with the Group Coordinator, preventing offset commits if the consumer was evicted during a rebalance.
• Kafka Streams: Enabling EOS requires a single configuration property:

  processing.guarantee=exactly_once_v2
  

---

4 Tricky Questions

1. Why was consumer.groupMetadata() added as an argument to sendOffsetsToTransaction() in Kafka 2.5 (KIP-447), and what critical vulnerability did it resolve?

Answer:

• In EOS v1, sendOffsetsToTransaction() accepted only the offset map and groupId.
• The Vulnerability: If a consumer thread hung on slow business logic, the Group Coordinator evicted it after max.poll.interval.ms and assigned its partition to a healthy pod. When the hung consumer finally unblocked, its transactional producer proceeded to commit offsets. Because the Transaction Coordinator was unaware that group membership had moved on, it accepted the commit, overwriting the new partition owner’s progress and corrupting consumer lag!
• ConsumerGroupMetadata provides the current generationId, memberId, and groupInstanceId.
• The Transaction Coordinator verifies these tokens against the Group Coordinator before executing the commit. If the member was fenced or rebalanced away, the transaction aborts with CommitFailedException or FencedInstanceIdException, preserving cluster integrity.

---

2. What happens if the broker configuration transaction.state.log.replication.factor is left at 1, and the node hosting the Transaction Coordinator crashes?

Answer:

• The internal topic __transaction_state stores all active transaction records, producer PID mappings, and epoch counters.
• If replication.factor = 1, losing that single broker renders transaction state completely inaccessible.
• Catastrophic Impact:
  • All producers coordinated by that broker fail immediately with CoordinatorNotAvailableException or TimeoutException.
  • In-flight transactions cannot be committed or aborted.
  • Consumers configured with isolation.level = read_committed freeze indefinitely at the Last Stable Offset (LSO) on all affected partitions, unable to consume subsequent messages from healthy producers!
• Production Requirement: Always configure transaction.state.log.replication.factor = 3 and transaction.state.log.min.isr = 2.

---

3. Why should developers never configure an excessively high transaction.timeout.ms (e.g., 1 hour)?

Answer:

• Consumers in read_committed mode read records strictly up to the Last Stable Offset (LSO)—the offset of the earliest uncommitted transaction.
• If a transactional producer opens a transaction and suffers an unhandled deadlock or network partition, the broker marks the transaction as ONGOING until transaction.timeout.ms expires.
• Throughout this entire duration, the LSO remains completely frozen. No read_committed consumer in the enterprise can advance past that offset, even if thousands of subsequent transactions from other producers commit successfully.
• Consumer lag explodes across all downstream systems. In production, configure transaction.timeout.ms = 10000..30000 (10–30 seconds) to ensure failed transactions are aborted rapidly.

---

4. Can repeated transaction rollbacks (producer.abortTransaction()) trigger memory leaks or disk bloat on Kafka brokers?

Answer:

• There is no JVM heap memory leak on the broker, but it causes disk and network bloat (Log Bloat):
• Kafka uses an append-only log architecture. When abortTransaction() is called, physical messages written to .log segment files are never removed or wiped! The broker merely appends an ABORT control marker at the tail of the log.
• Consumers with read_committed still fetch these aborted blocks across the network; the consumer client library inspects in-memory aborted transaction lists and discards the records client-side, burning network bandwidth and CPU cycles.
• Aborted data is purged from disk only when the entire segment file ages out according to time (retention.ms) or size (retention.bytes).

---

🎯 Interview Cheat Sheet

Core Concepts at a Glance

• Producer Checklist: transactional.id, enable.idempotence=true, acks=all.
• Consumer Checklist: isolation.level=read_committed, enable.auto.commit=false.
• Broker Checklist: transaction.state.log.replication.factor=3, transaction.state.log.min.isr=2.
• Read-Process-Write: Offset commits must go through producer.sendOffsetsToTransaction(offsets, groupMetadata).
• Zombie Fencing: Incrementing Producer Epoch prevents duplicate split-brain writes from revived pods.
• LSO Blocking: In-flight transactions freeze read_committed consumers until committed or aborted.
• Kafka Streams: Enable out of the box via processing.guarantee=exactly_once_v2.

---

Idempotent vs. Transactional Producer Comparison

Attribute	Idempotent Producer	Transactional Producer
Identifier	Ephemeral Producer ID (PID) (reset on reboot)	Persistent, static transactional.id
Scope	Single partition of a single topic	Multiple topics, partitions, and __consumer_offsets
Guarantees	De-duplicates network retry packets	Atomic multi-partition commit or rollback
Zombie Fencing	None	Full fencing via Producer Epoch
Complexity	Zero configuration (default in Kafka 3.0+)	Requires 2PC coordinator and transaction lifecycle code

---

Red Flags to Avoid

• ❌ “Enabling producer idempotence (enable.idempotence=true) gives full multi-partition Exactly-Once transactions.” (Idempotence is single-partition retry protection only; transactions require transactional.id).
• ❌ “In a transactional loop, the consumer commits offsets using consumer.commitSync().” (Breaks atomicity; offsets must be committed via producer.sendOffsetsToTransaction()).
• ❌ “Calling initTransactions() before every message sent.” (Must be invoked strictly once during producer startup).
• ❌ “Kafka transactions enable atomic commits to PostgreSQL databases.” (Kafka cannot coordinate 2PC with external relational databases).

---

Related Topics

• What Message Delivery Guarantees Does Kafka Provide — Delivery guarantee spectrum
• What is the Difference Between At-Most-Once, At-Least-Once and Exactly-Once — Semantic trade-offs
• What is Idempotent Producer — Idempotence mechanics
• What is Producer Acknowledgment and What Modes Exist (acks=0, 1, all) — Durability settings