Kafka Partitioned Log — Architecture, Replication, and Exactly-Once Semantics

By Marcus Vance · Lead Systems & Backend Architect · November 2026

Kafka Fundamentals — The Partitioned Log

Core Abstraction

Kafka is a **distributed, partitioned, replicated commit log**. Topics are split into **partitions** (ordered, immutable sequences of records). Each partition is a **segmented log file** on disk.

Record Structure

Record {
  key: bytes (nullable)      // Partitioning key
  value: bytes               // Payload
  timestamp: int64           // CreateTime / LogAppendTime
  headers: Header[]          // Key-value metadata
  offset: int64              // Position in partition (monotonic)
}

Topic → Partition → Segment

• Topic — Logical category (e.g., orders).
• Partition — Physical shard (e.g., orders-0, orders-1). Ordered, append-only.
• Segment — File on disk (00000000000000000000.log, .index, .timeindex, .snapshot). Default 1GB.

Offsets — Two Numbering Spaces

Each leader election increments **Leader Epoch**. Offsets reset per epoch. Prevents stale leader data.

Broker Architecture — Storage, Network, Request Pipeline

Storage Tiering (Kafka 3.6+ / Tiered Storage)

• Local tier — Broker disks (NVMe/SSD). Hot data. Fast reads/writes.
• Remote tier — S3, GCS, Azure Blob, HDFS. Cold segments. Cost-efficient.
• Segment migration — Based on remote.log.storage.enable, local.retention.bytes, local.retention.ms.

Request Pipeline

1. Acceptor — Accepts TCP connections (default port 9092).
2. Processor — Thread pool (num.network.threads, default 3). Reads requests, parses.
3. Request Handler — Thread pool (num.io.threads, default 8). Executes logic (produce, fetch, metadata).
4. Purgatory — Holds delayed requests (acks=all waiting for ISR, throttling).
5. Response Queue — Serializes responses back to processor.

Page Cache Reliance

Kafka **relies heavily on OS page cache**. Writes go to page cache (via mmap or FileChannel). Flush controlled by flush.messages / flush.ms (default: OS decides). **Zero-copy transfer** (sendfile) for consumer reads.

Replication — In-Sync Replicas (ISR)

Replication Factor & ISR

replication.factor (RF) = total replicas per partition. One **Leader**, rest **Followers**. **ISR** = replicas caught up to leader (within replica.lag.time.max.ms, default 30s).

ISR Protocol

1. Leader writes to local log.
2. Followers fetch via FetchRequest (high watermark = leader’s LEO).
3. Follower acks up to its LEO.
4. Leader updates ISR: if follower lag > threshold → **shrink ISR** (remove). If caught up → **expand ISR** (add).
5. Controller (KRaft) or ZooKeeper persists ISR changes to /brokers/topics//partitions//state.

acks Configuration — Durability vs Latency

Default: acks=all (since 3.0). Min ISR: min.insync.replicas (default 1). If ISR shrinks below min → producer gets NOT_ENOUGH_REPLICAS.

Leader Election — Controller / KRaft

**Controller** (single active) monitors broker heartbeats. On leader failure:

1. Controller detects (ZooKeeper watch / KRaft metadata log).
2. Selects new leader from **current ISR** (preferred: preferred.replica.election.enable).
3. Sends LeaderAndIsrRequest to new leader and followers.
4. New leader truncates log to high watermark, starts accepting writes.

**Unclean Leader Election** (unclean.leader.election.enable=false default): If no ISR, don’t elect out-of-sync replica (data loss). Enable only for availability-over-durability.

Consumer Groups — Cooperative Rebalancing

Group Protocol

Consumers join group via JoinGroup → SyncGroup. **Group Coordinator** (broker) assigns partitions.

Partition Assignment Strategies

Cooperative Rebalancing (Incremental)

1. Coordinator sends REVOKE only for partitions moving.
2. Consumer releases only those partitions, commits offsets.
3. Coordinator sends ASSIGN for new partitions.
4. Consumer starts fetching new partitions. **No full pause**.

Offset Management

• Auto-commit (enable.auto.commit=true, default): Periodic (5s) async commit. Risk: duplicate processing on crash.
• Manual commit (commitSync() / commitAsync()): Exact control. Pair with idempotent processing.
• Transaction commit (see below): Offsets committed atomically with output.

Consumer Lag Monitoring

# kafka-consumer-groups.sh --bootstrap-server broker:9092 --group my-group --describe
GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID
my-group        orders          0          1000            1050            50              consumer-1
my-group        orders          1          995             1040            45              consumer-2

Exactly-Once Semantics — Transactions

Problem: Duplicate Writes

Producer retries (network timeout) → broker may have written but ack lost → duplicate. Consumer reprocesses on rebalance → duplicate.

Solution: Idempotent Producer + Transactions

Idempotent Producer (enable.idempotence=true)

• Producer gets **PID** (Producer ID) and **sequence number** per partition.
• Broker tracks (PID, partition, seq) — deduplicates on retry.
• Requires acks=all, max.in.flight.requests.per.connection=5 (default).
• **Per-partition deduplication only** — not cross-partition.

Transactions — Atomic Multi-Partition Write

producer.initTransactions()
producer.beginTransaction()
producer.send(new ProducerRecord("orders", key, value))
producer.send(new ProducerRecord("inventory", key, value))
producer.sendOffsetsToTransaction(
  Map.of(new TopicPartition("orders", 0), new OffsetAndMetadata(1001)),
  "consumer-group-id"
)
producer.commitTransaction()  // or abortTransaction()

Transaction Log — __transaction_state

• Internal topic (50 partitions default). Stores transaction state: Ongoing, PrepareCommit, Commit, Abort.
• Coordinator (broker) manages transaction lifecycle.
• **Two-phase commit**: Producer sends PrepareCommit → Coordinator writes to transaction log → Coordinator writes Commit → Producer receives ack.
• On failure: Coordinator aborts (writes Abort). Consumers see Abort marker → skip.

Transactional Consumer — Read Committed

props.put("isolation.level", "read_committed")  // Default: read_uncommitted
// Only returns records from committed transactions.
// Aborted transaction records filtered out (consumer sees control records).

Exactly-Once End-to-End

1. Producer: enable.idempotence=true + Transactions.
2. Consumer: isolation.level=read_committed + manual offset commit inside transaction (sendOffsetsToTransaction).
3. Downstream: Idempotent writes (deduplication key) or transactional sink.

KRaft Mode — Kafka Without ZooKeeper

Architecture

Metadata (topics, partitions, ISR, configs, ACLs) stored in **Kafka’s own Raft-based metadata log** (__cluster_metadata topic). **Controller** = Raft leader. **Brokers** = Raft followers (voters) or observers.

Migration Path

1. Enable KRaft in ZooKeeper cluster: metadata.log.dir, controller.quorum.voters.
2. Start controller quorum (3+ nodes).
3. Migrate brokers: process.roles=broker (or broker,controller for combined).
4. Remove ZooKeeper.

Benefits

• Single system — no ZooKeeper ops, version skew, split-brain.
• Faster controller failover (Raft election ~ms vs ZK ~seconds).
• Scalable metadata (millions of partitions).
• Simpler security (single auth system).

Production Operations & Tuning

Key Broker Configs

# Network / Threads
num.network.threads=8
num.io.threads=16
socket.request.max.bytes=104857600  # 100MB

# Log / Retention
log.dirs=/mnt/kafka/data1,/mnt/kafka/data2  # Multiple dirs = parallel I/O
log.segment.bytes=1073741824  # 1GB segments
log.retention.hours=168       # 7 days
log.retention.bytes=-1        # No size limit
log.roll.hours=168

# Replication
replica.fetch.max.bytes=1048576
replica.fetch.wait.max.ms=500
replica.lag.time.max.ms=30000

# Flush (rely on page cache mostly)
flush.messages=10000
flush.ms=1000

# Compression (producer side default)
compression.type=snappy  # or zstd for better ratio

# Quotas (prevent noisy neighbors)
quota.window.size.seconds=1
quota.window.num=10

Monitoring — Critical Metrics

Operational Procedures

Rolling Restart

1. Set controlled.shutdown.enable=true (default).
2. Stop broker → controller moves partitions (preferred replica election).
3. Start broker → rejoins ISR, catches up.
4. One at a time. Monitor UnderReplicatedPartitions = 0 between restarts.

Partition Reassignment (Rebalance)

# Generate reassignment plan
kafka-reassign-partitions.sh --bootstrap-server broker:9092 
  --topics-to-move-json-file topics.json 
  --broker-list "1,2,3" --generate

# Execute
kafka-reassign-partitions.sh --bootstrap-server broker:9092 
  --reassignment-json-file reassignment.json --execute

# Verify
kafka-reassign-partitions.sh --bootstrap-server broker:9092 
  --reassignment-json-file reassignment.json --verify

Log Compaction (Key-Based Retention)

# Topic config
kafka-configs.sh --bootstrap-server broker:9092 
  --entity-type topics --entity-name user-profiles 
  --alter --add-config cleanup.policy=compact,min.cleanable.dirty.ratio=0.5

Keeps latest value per key. Used for changelog, materialized views, KTables.

TL;DR — Key Takeaways

• Kafka = **Partitioned, replicated commit log**. Partitions = unit of parallelism & ordering.
• ISR = in-sync replicas. acks=all waits for full ISR. min.insync.replicas enforces durability.
• CooperativeSticky assignor = incremental rebalances (no stop-the-world).
• Transactions = idempotent producer + transaction log + read_committed consumer = exactly-once.
• KRaft = native Raft metadata, no ZooKeeper. Standard since 3.0.
• Tiered Storage = local NVMe + remote S3. Infinite retention at low cost.
• Monitor: UnderReplicatedPartitions, OfflinePartitions, ConsumerLag, ControllerCount.

Next in series: Linux RCU Synchronization — Read-Copy-Update internals and usage patterns.