Delivery Guarantees & Reliability

flowchart TB
  subgraph producer["Producer layer"]
    P1[acks / retries / idempotence]
  end
  subgraph broker["Broker layer"]
    B1[replication / ISR / min.insync.replicas]
  end
  subgraph consumer["Consumer layer"]
    C1[offset commit timing / idempotent handler]
  end
  P1 --> B1 --> C1

At-most-once

Definition: a record may be lost; it will not be delivered twice. Typically: acks=0 (fire-and-forget), or commit offsets before processing with no retry on failure.

Where it is acceptable:

• Metrics and telemetry where statistical sampling tolerates gaps (dashboard approximations)
• High-volume click streams where 0.1% loss is cheaper than 2× infrastructure for durability
• Ephemeral notifications (live presence pings) that self-heal on next event
• Debug/logging pipelines—not audit, billing, or inventory

acks=0
retries=0
enable.idempotence=false

At-least-once

Definition: a record will not be lost under retry scenarios; it may be delivered more than once. This is the default for most production Kafka systems: acks=all or acks=1 + retries > 0 on producer, process-then-commit (or auto-commit risks) on consumer.

Requirement: consumers and downstream sinks must be idempotent—same message applied twice produces the same state. Patterns: natural keys in DB upserts, dedup table in Redis/PostgreSQL, idempotency keys in event payload.

for (ConsumerRecord<String, OrderEvent> record : records) {
  String idempotencyKey = record.topic() + "-" + record.partition() + "-" + record.offset();

  if (dedupStore.exists(idempotencyKey)) {
    continue;  // already processed this exact broker record
  }

  processOrder(record.value());
  dedupStore.put(idempotencyKey, Duration.ofDays(7));
  consumer.commitSync(Map.of(
      new TopicPartition(record.topic(), record.partition()),
      new OffsetAndMetadata(record.offset() + 1)
  ));
}

Exactly-once

Definition: each record’s effect appears exactly once in downstream state—no loss, no duplicate side effects. Kafka does not magically guarantee this for your database writes; it provides building blocks. Two approaches:

Approach 1: Idempotent consumer (application-level)

Deduplicate using message ID in external store (Redis, DB unique constraint, DynamoDB conditional write). Works with any broker config; you own TTL, storage cost, and replay semantics. Still at-least-once on the wire—exactly-once is an application invariant.

Approach 2: Kafka transactions (platform-level)

Atomic consume → process → produce with offset commit tied to output records in one transaction. Kafka Streams processing.guarantee=exactly_once_v2 uses this internally. Requires transactional producer, read_committed consumers, and compatible brokers.

transactional.id

Stable string per producer instance role (e.g. inventory-processor-pod-7). Broker maps it to a Producer ID (PID) and epoch. When a new producer registers with the same transactional.id, epoch increments and fences the old producer— zombie instances get ProducerFencedException on next operation.

Transaction coordinator

One broker acts as transaction coordinator per transactional.id (hash of id → broker). Manages transaction state machine: Ongoing → PrepareCommit → CompleteCommit (or Abort). Not the same as partition leader—dedicated coordination role.

__transaction_state internal topic

Compact internal topic (default 50 partitions, RF=3) storing transaction metadata: transactional id, producer id, epoch, status. Survives coordinator failover. Never manually delete in production.

Transaction phases

1. initTransactions() — register PID, recover pending transactions, fence zombies
2. beginTransaction() — start new txn; records tagged with transactional markers
3. send() — batches include PID, epoch, sequence; written to partition logs as uncommitted
4. sendOffsetsToTransaction(Map, groupMetadata) — atomically commit consumer offsets with output records
5. commitTransaction() — write COMMIT marker; records become visible to read_committed
6. abortTransaction() — write ABORT marker; records filtered out for consumers

sequenceDiagram
  participant App as Application
  participant TC as Transaction Coordinator
  participant L as Partition Leaders
  participant C as read_committed Consumer

  App->>TC: initTransactions()
  App->>TC: beginTransaction()
  App->>L: produce (uncommitted data)
  App->>TC: sendOffsetsToTransaction
  App->>TC: commitTransaction()
  TC->>L: COMMIT control record
  C->>L: fetch — sees committed data only

Partition log timeline:
[txn-start][data][data][ABORT]     ← aborted txn — invisible to read_committed
[txn-start][data][data][COMMIT]    ← committed txn — visible
         ↑ uncommitted until COMMIT marker appended

read_committed isolation

Consumers with isolation.level=read_committed filter aborted transactions and delay visibility of open transactions until commit marker arrives—prevents consumers from seeing partial writes. Default read_uncommitted sees all records immediately (including aborted txn data before abort marker—ordering nuance).

producer.initTransactions();

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

  try {
    producer.beginTransaction();

    for (ConsumerRecord<String, String> record : records) {
      String output = transform(record.value());
      producer.send(new ProducerRecord<>("output-topic", record.key(), output));
    }

    Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
    for (TopicPartition tp : records.partitions()) {
      List<ConsumerRecord<String, String>> part = records.records(tp);
      long last = part.get(part.size() - 1).offset();
      offsets.put(tp, new OffsetAndMetadata(last + 1));
    }

    producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
    producer.commitTransaction();
  } catch (Exception e) {
    producer.abortTransaction();
    throw e;
  }
}

spring.kafka:
  producer:
    transaction-id-prefix: inventory-txn-
    properties:
      enable.idempotence: true
      acks: all
  consumer:
    enable-auto-commit: false
    properties:
      isolation.level: read_committed

Performance cost

Transactions add coordination round-trips per commit, control records on partitions, and coordinator load. Typical overhead: ~3–5% throughput reduction and higher p99 latency per commit boundary compared to idempotent at-least-once. Batch more records per transaction where business logic allows.

Replication factor recommendations

• Production: replication.factor=3 — survives single broker loss with quorum
• Never < 2 in any environment that cares about data—single replica is a single disk failure from loss
• Development: RF=1 acceptable on single-broker Docker; not representative of production behavior

The safe production triad

# Broker defaults
default.replication.factor=3
min.insync.replicas=2
unclean.leader.election.enable=false

# Per-topic override (kafka-configs.sh)
min.insync.replicas=2
cleanup.policy=delete

What happens when a broker fails

1. Leader broker dies — controller detects via ZK/KRaft session; partitions with leader on dead broker become unavailable for produce/fetch briefly
2. Leader election — controller elects new leader from ISR only (if unclean disabled)
3. Followers catch up — new leader serves; out-of-sync replicas truncate to HW and replicate
4. Clients refresh metadata — producers/consumers redirect to new leaders (automatic metadata refresh)
5. Under-replicated partitions — metric rises until followers rejoin ISR

sequenceDiagram
  participant P as Producer
  participant B1 as Broker 1 Leader
  participant B2 as Broker 2 Follower
  participant C as Controller
  participant B3 as Broker 3 Follower

  P->>B1: produce acks=all
  B1->>B2: replicate
  B1->>B3: replicate
  Note over B1: Broker 1 crashes
  C->>B2: elect leader for partition
  P->>B2: produce resumes after metadata refresh

Preferred leader election

Each partition has a preferred leader (replica 0 in assignment). After broker restart, kafka-leader-election.sh --election-type preferred or auto.leader.rebalance.enable (legacy) moves leadership back to preferred replica— rebalancing load to intended topology and avoiding stranded leaders on replacement brokers.

Unclean leader election

unclean.leader.election.enable=false (production default): never elect a leader from outside ISR. Partition may go offline rather than serve stale data. true: availability over consistency—may promote lagging replica and lose unreplicated records.

Producer ID (PID)

On first connection with enable.idempotence=true, broker assigns a unique Producer ID (64-bit). All batches from this producer session carry this PID. Broker uses PID + partition + sequence to deduplicate.

Epoch

When a producer with transactional.id calls initTransactions(), broker increments epoch. Any older producer instance with same transactional id but lower epoch receives ProducerFencedException—prevents zombie writers after failover or rolling deploy.

Sequence number

Per-partition monotonic counter starting at 0 for each PID epoch. Broker rejects:

• Duplicate sequence — retry of already-committed batch (silent success, no duplicate in log)
• Out-of-order sequence — gap detected (e.g. max.in.flight > 1 without idempotence caused reorder); with idempotence, broker enforces strict order per partition

PID=7021, Partition orders-0:
  seq=0  batch [A,B]   → accepted
  seq=1  batch [C]     → network timeout, producer retries
  seq=1  batch [C]     → duplicate → broker acks without rewrite
  seq=2  batch [D]     → accepted

Automatic config coupling

When enable.idempotence=true, client enforces:

• acks=all
• retries > 0 (Integer.MAX_VALUE)
• max.in.flight.requests.per.connection ≤ 5 (and safe with 5)

Limitations

• Deduplication only within a producer session (same PID)
• Producer restart without transactional.id → new PID → retries across sessions can duplicate
• Does not deduplicate logically identical messages sent with different sequence contexts (two explicit sends)
• No cross-partition atomicity—that requires transactions

enable.idempotence=true
# Implicit: acks=all, retries=MAX, max.in.flight=5
# No transactional.id — no consume-transform-produce atomicity

The full Kafka-native pattern

1. Transactional producer with stable transactional.id
2. Consumer with enable.auto.commit=false and isolation.level=read_committed
3. Process records from poll batch
4. Produce output records in same transaction
5. sendOffsetsToTransaction — offset commit atomic with produces
6. commitTransaction — all visible together or none

flowchart LR
  IN[input topic]
  CON[Consumer poll]
  TXN[Transactional producer]
  OUT[output topic]
  OFF[__consumer_offsets]
  IN --> CON --> TXN
  TXN --> OUT
  TXN --> OFF

Kafka Streams exactly_once_v2

processing.guarantee=exactly_once_v2 (EOS v2, Kafka 2.5+) wraps the same primitives: transactional producer per stream thread, changelog topics for state stores, offset commits tied to output and state updates in one transaction boundary.

application.id=inventory-aggregator
processing.guarantee=exactly_once_v2
replication.factor=3
# Creates internal changelog + repartition topics with RF=3

External side effects — the honesty section

Kafka EOS does not roll back your database. Patterns for external writes:

Cost — when to enable EOS

• Enable when duplicate output records cause financial errors, double inventory deduction, or incorrect aggregates
• Skip when idempotent sinks are cheap (upsert by key) and 1–2% duplicate rate is harmless
• Measure throughput and p99 before/after—expect 3–10% hit depending on commit frequency and partition count

# Producer
enable.idempotence: true
transactional.id: ${APP_NAME}-${INSTANCE_ID}
acks: all

# Consumer
enable.auto.commit: false
isolation.level: read_committed

# Broker / topic
replication.factor: 3
min.insync.replicas: 2
unclean.leader.election.enable: false

# Streams (if applicable)
processing.guarantee: exactly_once_v2