60-Day Kafka 4 Learning Plan · Week 2 — Day 11 of 60

60-Day Kafka 4 Learning Plan · Week 2 — Kafka Internals Sources: Kafka: The Definitive Guide Ch.6 · kafka.apache.org/43

Goal

Understand the three delivery guarantees Kafka supports, how the idempotent producer eliminates duplicates, how transactions enable atomic multi-partition writes, how zombie producers get fenced, and how to configure, monitor, and test end-to-end exactly-once semantics in Spring Boot.

1. The three delivery guarantees

At-most-once

0 or 1 delivery — message may be lost, never duplicated.
Send and forget: no retry, no ACK wait.
Config: acks=0, retries=0
Use for: non-critical metrics, click logs, telemetry.

At-least-once (default)

1 or more deliveries — never lost, may be duplicated on retry.
Producer retries on failure → broker may receive the same message twice.
Config: acks=all, retries > 0
Consumer must be idempotent (handle duplicates gracefully).

Exactly-once ★

Exactly 1 delivery — no loss, no duplicates.
Hardest to achieve; requires idempotent producer + Kafka transactions.
Config: enable.idempotence=true + transactional.id

2. Idempotent producer

Problem: on retry, the broker may receive the same message twice if the ACK was lost in transit — causing silent duplicates even with retries > 0.

Solution: each producer session gets a unique PID (Producer ID). Each message gets a monotonic sequence number per (PID, partition) pair.

Producer (PID=42, seq=5) ──► Broker (stores seq=5 ✓)
                         ◄── ACK lost!
Producer retries (seq=5) ──► Broker: seq=5 already seen → discard silently ✓

Limitations:

Deduplicates within a single producer session on a single partition only.
Does not span multiple topics or across producer restarts.
For cross-partition exactly-once, use transactions.

Enable:

spring:
  kafka:
    producer:
      enable-idempotence: true

This automatically sets acks=all, retries=Integer.MAX_VALUE, and max.in.flight.requests.per.connection=5.

3. Kafka transactions — atomic multi-partition writes

Transactions allow a producer to write to multiple partitions (and commit consumer offsets) atomically — all succeed or all roll back. Required for Kafka Streams EOS.

Transaction flow

producer.initTransactions();        // 1. register transactional.id with broker

producer.beginTransaction();        // 2. start transaction

producer.send("topic-A", msgA);     // 3. buffered writes
producer.send("topic-B", msgB);

producer.sendOffsetsToTransaction(  // 4. atomic offset commit (optional)
    offsets, groupMetadata);

// 5a. success path:
producer.commitTransaction();       // ✓ all writes + offsets committed atomically

// 5b. error path:
producer.abortTransaction();        // ✗ all writes rolled backHow it works internally:

How it works internally:

A Transaction Coordinator broker tracks the transaction state.
Writes are held until commitTransaction() — then a commit marker is appended to all affected partitions.
Consumers with isolation.level=read_committed wait for the marker before delivering messages.

4. Consumer isolation.level

Transactions only matter if consumers honour the commit/abort markers.

spring:
  kafka:
    consumer:
      isolation-level: read_committed

Caveat with compaction: on a topic that’s both transactional and log.cleanup.policy=compact, aborted-transaction records still occupy disk until compacted, and commit/abort markers themselves are cleaned up separately (transaction.remove.expired.transaction.cleanup.interval.ms). Don’t assume read_committed means aborted data disappears instantly from disk — it only means consumers don’t see it.

5. Zombie fencing — how `initTransactions()` prevents split-brain writes

A zombie producer is an old instance of your app that’s still alive (e.g. a slow shutdown, a network partition that isolated it, a stuck GC pause) while a new instance has already started with the same transactional.id. Without fencing, both could write conflicting data.

Instance A starts → transactional.id="order-svc-1" → epoch=5 (assigned by coordinator)
Instance A hangs (GC pause, thinks it's still leader)

Instance B restarts with SAME transactional.id="order-svc-1"
  → calls initTransactions() → coordinator bumps epoch to 6
  → coordinator now REJECTS any write tagged with epoch 5

Instance A resumes, tries to commit with epoch=5 → broker rejects: ProducerFencedException

Key mechanics:

Every call to initTransactions() increments the producer epoch for that transactional.id.
The broker only accepts writes carrying the current epoch — any zombie holding a stale epoch is automatically fenced out.
This is why transactional.id must be stable and unique per logical producer instance (e.g. order-svc-${pod-name} in Kubernetes) — reusing the same ID across genuinely different instances is exactly what fencing is designed to protect against, and rotating it randomly on every restart defeats the purpose of transactions surviving a crash.

6. Transaction timeout & performance cost

transaction.timeout.ms=60000   # default — abort if not committed within 60s
transaction.max.timeout.ms=900000  # broker-side ceiling on what producers can request

If a transaction is left open too long (application bug, deadlock, slow downstream call before commitTransaction()), the coordinator proactively aborts it once transaction.timeout.ms elapses — otherwise a stuck producer would hold locks on partitions indefinitely, blocking read_committed consumers from proceeding past that point.

Performance trade-offs to plan for:

Every transaction requires at least 2 extra broker round-trips (coordinator registration + commit marker write) beyond a plain idempotent send — expect meaningfully lower throughput than non-transactional idempotent producers under high message-per-second workloads.
read_committed consumers must buffer records until they see the commit/abort marker, which adds latency proportional to transaction duration — long-running transactions directly slow down consumer-side delivery, not just producer-side commits.
Keep transactions short and focused (ideally a single consume-transform-produce cycle) rather than batching many unrelated business operations into one transaction.

7. Error handling in transactional `@KafkaListener`s

When @Transactional wraps a @KafkaListener, an exception thrown inside the method triggers abortTransaction() automatically — none of the sends inside that invocation are committed, and Spring’s container will redeliver the record on the next poll (subject to your DefaultErrorHandler retry/backoff policy from Day 6).

@Transactional
@KafkaListener(topics = "orders", groupId = "order-svc")
public void process(ConsumerRecord<String, Order> record) {
    Order order = record.value();
    if (order.amount() < 0) {
        throw new IllegalArgumentException("Negative amount"); // triggers abort, no partial writes
    }
    kafkaTemplate.send("invoices", order.id(), toInvoice(order));
}

Important: this abort-on-exception behavior means a permanently failing record (a poison pill) will loop forever without a DefaultErrorHandler + DLT strategy — the transaction protects atomicity, but it doesn’t protect you from infinite reprocessing of a message that will never succeed.

8. Monitoring transactions

# Inspect open/pending transactions for debugging
kafka-transactions.sh --bootstrap-server localhost:9092 --list

# Force-abort a stuck transaction (last resort, use with care)
kafka-transactions.sh --bootstrap-server localhost:9092 \
  --abort --topic orders --partition 0 --producer-id 42 --producer-epoch 6

9. Testing exactly-once pipelines

@SpringBootTest
@EmbeddedKafka(partitions = 3, topics = {"orders", "invoices"})
class OrderConsumerEosTest {

    @Autowired KafkaTemplate<String, Order> kafkaTemplate;
    @Autowired ConsumerFactory<String, Invoice> consumerFactory;

    @Test
    void publishesInvoiceExactlyOnceOnSuccess() {
        kafkaTemplate.send("orders", "o1", new Order("o1", 100));

        var invoiceConsumer = consumerFactory.createConsumer("verify-group", "test");
        invoiceConsumer.subscribe(List.of("invoices"));

        var records = KafkaTestUtils.getRecords(invoiceConsumer, Duration.ofSeconds(5));
        assertThat(records.count()).isEqualTo(1); // exactly one invoice, no duplicates
    }

    @Test
    void abortsTransactionOnNegativeAmount() {
        kafkaTemplate.send("orders", "o2", new Order("o2", -50));
        // assert invoices topic receives NOTHING for o2, and offset is retried/DLT'd per error handler
    }
}

Set the test consumer’s isolation.level=read_committed explicitly — otherwise @EmbeddedKafka tests can pass locally while missing real dirty-read bugs that only show up against a broker respecting transaction boundaries.

10. Common pitfalls

Random/rotating transactional.id per restart — defeats zombie fencing’s purpose; use a stable ID tied to the logical instance (pod name, partition assignment), not a UUID generated fresh each boot
Long business logic inside a transaction — slow external calls (HTTP, DB) between beginTransaction() and commitTransaction() hold locks and delay every read_committed consumer downstream
Assuming transactions protect against poison pills — they guarantee atomicity, not liveness; still need DefaultErrorHandler + DLT for permanently-failing records
Forgetting isolation.level=read_committed on the consumer — the whole transactional setup is silently pointless if consumers default to read_uncommitted and see uncommitted/aborted data anyway
Treating transactions as free — expect real throughput cost vs plain idempotent producers; don’t reach for transactions when idempotence alone (single-partition dedup) already meets the requirement

11. End-to-end exactly-once (EOS) checklist

Producer side

enable.idempotence=true
transactional.id=unique-app-id      # unique per application instance
acks=all                            # auto-set by idempotence
retries=2147483647                  # auto-set by idempotence
max.in.flight.requests.per.connection=5

Consumer side

isolation.level=read_committed
enable.auto.commit=false
# Use sendOffsetsToTransaction() — NOT commitSync() — inside the transaction

12. Spring Boot: EOS config + @Transactional

spring:
  kafka:
    producer:
      transaction-id-prefix: tx-order-svc
      acks: all
      enable-idempotence: true
    consumer:
      isolation-level: read_committed
      enable-auto-commit: false

@Service
@Slf4j
@RequiredArgsConstructor
public class OrderConsumer {

    private final KafkaTemplate<String, Invoice> kafkaTemplate;

    @Transactional                    // Spring wraps in Kafka transaction
    @KafkaListener(topics = "orders", groupId = "order-svc")
    public void process(ConsumerRecord<String, Order> record) {
        Order order = record.value();
        log.info("Processing order={}", order.id());
        kafkaTemplate.send("invoices", order.id(), toInvoice(order));
        // offset committed atomically with the send — exactly-once
    }

    private Invoice toInvoice(Order o) {
        return new Invoice(o.id(), o.amount());
    }
}

Spring’s KafkaTransactionManager (auto-configured when transaction-id-prefix is set) handles initTransactions, beginTransaction, and commitTransaction / abortTransaction around each @KafkaListener invocation.

13. Comparison table

Key Takeaways

At-least-once is the default — the consumer must be idempotent to tolerate duplicates
The idempotent producer uses PID + sequence numbers to deduplicate retries per partition
Transactions group writes to multiple partitions atomically — all commit or all abort
Zombie fencing via producer epoch bumps on initTransactions() is what makes transactional.id stability critical, not optional
Transactions have a real performance cost — extra broker round-trips and consumer-side buffering — so keep them short
Transactions guarantee atomicity, not liveness — poison pills still need DefaultErrorHandler + DLT
The consumer must set isolation.level=read_committed to honour transaction boundaries
Spring Boot: transaction-id-prefix + @Transactional on @KafkaListener = the EOS pattern
enable-idempotence=true automatically configures acks=all and retries=MAX

Support me through GitHub Sponsors.

➡️ Day 12: Keys & Partitioning

Resources

📘 Kafka: The Definitive Guide — Chapter 6 (Reliable Data Delivery)
🌐 kafka.apache.org/43/documentation — transactions & idempotence
🌐 Spring Kafka — KafkaTransactionManager
🌐 KIP-98: Exactly Once Delivery and Transactional Messaging

👉 Link to Medium blog