Source: kafka · kafka.md · updated 2026-08-07 · 🔒 secret gist

Synced verbatim from gist.github.com/bl9.

Apache Kafka Internals — A Maintainer's Deep Dive

Source tree: github.com/apache/kafka @ 930ebc5608bb0ac938085321d09b402b850ca87b Version: 4.4.0-SNAPSHOT · Scala 2.13.18 · LATEST_PRODUCTION = IBP_4_3_IV0 Fetched: 2026-08-06

Every file path, class name, and line-level behaviour below was read out of that tree. Where a design decision has a rationale that only exists in a code comment, the comment is the citation — Kafka's best documentation is in-source and most of it never made it into the website docs.

Table of Contents

1. How to read this codebase
2. Module topology
3. The log: on-disk format
   • 3.1 Directory and file layout
   • 3.2 RecordBatch v2, byte by byte
   • 3.3 The inner Record
   • 3.4 Why the CRC is where it is
   • 3.5 The offset index
   • 3.6 The warm-section binary search
   • 3.7 Time index, txn index, epoch cache, producer snapshots
4. The write path, end to end
   • 4.1 Acceptor / Processor / RequestChannel
   • 4.2 Request handler pool and callback re-entry
   • 4.3 ReplicaManager → Partition → UnifiedLog
   • 4.4 Anatomy of UnifiedLog.append
   • 4.5 LogValidator: offset assignment and recompression
   • 4.6 Segment roll
   • 4.7 The fsync question
   • 4.8 Idempotence: ProducerStateManager
   • 4.9 acks=all and the purgatory
5. The read path
   • 5.1 Offset → file position
   • 5.2 Zero-copy and where it breaks
   • 5.3 Fetch sessions (KIP-227)
   • 5.4 DelayedFetch
   • 5.5 Read isolation: HW vs LSO
   • 5.6 Tiered storage reads
6. Hierarchical timing wheels and the purgatory
7. Replication
   • 7.1 Pull replication, and why
   • 7.2 The high watermark, and its one-round-trip lag
   • 7.3 ISR: expand, shrink, maximal ISR
   • 7.4 Leader epochs and log reconciliation
   • 7.5 Eligible Leader Replicas (KIP-966)
8. KRaft
   • 8.1 How it differs from textbook Raft
   • 8.2 The quorum state machine, including Prospective
   • 8.3 Leader HW computation
   • 8.4 Snapshots
   • 8.5 QuorumController: a single-threaded event loop
   • 8.6 MetadataImage / MetadataDelta / MetadataLoader
   • 8.7 Feature flags and MetadataVersion
9. Coordinators
   • 9.1 The coordinator runtime
   • 9.2 Group coordinator: KIP-848
   • 9.3 The reconciliation state machine
   • 9.4 Assignors
   • 9.5 Transaction coordinator and KIP-890
   • 9.6 Share groups (KIP-932)
10. Log compaction
11. Producer client internals
12. Consumer client internals
13. Quotas, throttling, and backpressure
14. Comparison with other streaming systems
15. Design lessons: building your own
16. Maintainer's appendix

1. How to read this codebase

A few navigational facts that will save you a week.

• Scala is nearly gone from the hot paths. The log layer, the indexes, the compactor, the coordinators, the Raft implementation, the purgatory, and the timing wheel are all Java now. What remains in Scala is essentially: KafkaApis, ReplicaManager, Partition, SocketServer, the fetcher threads, KafkaConfig, and the transaction coordinator. The migration is mechanical and ongoing; don't assume a class is where a 2021 blog post says it is.
• The module boundary is a dependency boundary, enforced. clients cannot depend on server. storage cannot depend on core. This is why you see interfaces like RemoteStorageManager and AlterPartitionManager sitting in low modules with the implementation in core. When you add a class, the module you pick determines what you're allowed to call.
• internals packages are the real API. org.apache.kafka.storage.internals.log is where the log lives. org.apache.kafka.common.record.internal is where the record format lives (it moved under internal recently — a lot of stale documentation points at org.apache.kafka.common.record.DefaultRecordBatch, which no longer exists).
• Comments are load-bearing. AbstractIndex has a 40-line comment explaining a page-cache argument that is the entire reason the class is not a plain binary search. DelayedOperationPurgatory has a comment enumerating a specific 7-step deadlock. These are not decoration; they encode post-mortems.
• The generator module is real code. generator/ compiles the JSON schemas in clients/src/main/resources/common/message/*.json into request/response classes. If you're adding an RPC field, you edit JSON, not Java. ./gradlew processMessages regenerates.

2. Module topology

clients/                 protocol, record format, producer, consumer, admin, serde, network
                         (the only artifact most users depend on)
raft/                    KafkaRaftClient — the KRaft consensus implementation
metadata/                controller (QuorumController), metadata records, image/delta, loader
server-common/           purgatory, timing wheel, MetadataVersion, shared utils
server/                  broker-side pieces that don't need core's Scala: quotas, fetch sessions
storage/                 UnifiedLog, LogSegment, indexes, compaction, tiered storage
coordinator-common/      the generic coordinator runtime (event loop + replicated state machine)
group-coordinator/       consumer groups, share groups, streams groups
share-coordinator/       durable share-group acknowledgement state
transaction-coordinator/ transaction metadata types (state machine still partly Scala in core)
core/                    KafkaApis, ReplicaManager, Partition, SocketServer, BrokerServer,
                         ControllerServer, fetcher threads
streams/                 Kafka Streams
connect/                 Kafka Connect
tools/, shell/, trogdor/ CLI, metadata shell, fault injection harness
jmh-benchmarks/          microbenchmarks — read these, they document the hot paths

The dependency direction is roughly clients ← server-common ← {raft, storage, server} ← {metadata, coordinator-common} ← core. core is the only module that can see everything, which is why it's the residual Scala pile.

3. The log: on-disk format

3.1 Directory and file layout

A log.dirs entry contains one directory per partition, named <topic>-<partition>:

/var/lib/kafka/data/
├── meta.properties                  # cluster id, node id, directory id
├── orders-0/
│   ├── partition.metadata           # topic id (written before any data — see below)
│   ├── leader-epoch-checkpoint      # (epoch, startOffset) pairs
│   ├── 00000000000000000000.log     # segment: the actual records
│   ├── 00000000000000000000.index   # offset index, mmapped, sparse
│   ├── 00000000000000000000.timeindex
│   ├── 00000000000000000000.snapshot # producer state snapshot at this base offset
│   ├── 00000000000000368142.log
│   ├── 00000000000000368142.index
│   ├── 00000000000000368142.timeindex
│   ├── 00000000000000368142.txnindex # only if aborted txns exist in this segment
│   └── 00000000000000368142.snapshot
└── __cluster_metadata-0/            # the KRaft log, on controllers and brokers alike
    ├── 00000000000000000000.log
    └── ...-0000000000-0000000000.checkpoint   # KRaft snapshots

The filename is the base offset, zero-padded to 20 digits, which makes lexical sort equal numeric sort. Everything is derivable from that number, which is why LogSegments can be a ConcurrentNavigableMap<Long, LogSegment> and segment lookup is a floorEntry.

One subtlety that costs people data: UnifiedLog.append opens with

// We want to ensure the partition metadata file is written to the log dir before any log data is written to disk.
// This will ensure that any log data can be recovered with the correct topic ID in the case of failure.
maybeFlushMetadataFile();

partition.metadata must hit disk before the first record. Without it, a crash leaves a directory of records that can't be attributed to a topic ID, and topic IDs are how KRaft distinguishes a recreated topic from the original.

Transient files you will see and should recognize:

.cleaned vs .swap is the crash-consistency protocol for compaction: the presence of .swap means the new segment is fully written and the rename is idempotent to redo.

3.2 RecordBatch v2, byte by byte

From clients/src/main/java/org/apache/kafka/common/record/internal/DefaultRecordBatch.java:

Offset  Size  Field
------  ----  -----------------------------------------------------
  0      8    BaseOffset            int64
  8      4    Length                int32   (bytes after this field)
 12      4    PartitionLeaderEpoch  int32   ← NOT covered by CRC
 16      1    Magic                 int8    (= 2)
 17      4    CRC                   uint32  CRC-32C over bytes 21..end
 21      2    Attributes            int16
 23      4    LastOffsetDelta       int32   (also = LastSequenceDelta)
 27      8    BaseTimestamp         int64
 35      8    MaxTimestamp          int64
 43      8    ProducerId            int64
 51      2    ProducerEpoch         int16
 53      4    BaseSequence          int32
 57      4    RecordsCount          int32
 61      -    Records               [Record]  (compressed as one blob if enabled)

RECORD_BATCH_OVERHEAD = 61 bytes. LOG_OVERHEAD = 12 (BaseOffset + Length) — that's the framing prefix that lets you skip a batch without parsing it.

Attributes bitfield:

bit  0-2  compression type (0 none, 1 gzip, 2 snappy, 3 lz4, 4 zstd)
bit  3    timestamp type (0 CreateTime, 1 LogAppendTime)
bit  4    transactional
bit  5    control batch (contains markers, not user data)
bit  6    delete horizon set (compaction tombstone retention)
bit 7-15  unused

Three properties of this layout matter enormously and are worth stealing:

1. Compression is per-batch, not per-record. The compressed blob starts right after RecordsCount. This means the broker can route, index, and replicate a batch without decompressing it. Systems that compress per-message (or per-request-with-per-message framing) give up an order of magnitude on wire efficiency for high-cardinality small records.
2. Offsets and timestamps are deltas. A record stores OffsetDelta and TimestampDelta as varints relative to the batch header. A batch of 10,000 sequential records with clock-adjacent timestamps costs ~1–2 bytes per record for both fields combined.
3. The batch is the unit of everything: the unit of compression, of CRC, of idempotence (BaseSequence/LastOffsetDelta), of transactional membership, and of replication. The entire system's constant factors come from having chosen the batch as the atom.

3.3 The inner Record

From DefaultRecord.java:

Record =>
  Length          Varint
  Attributes      Int8      (currently all bits unused)
  TimestampDelta  Varlong
  OffsetDelta     Varint
  KeyLength       Varint    (-1 = null)
  Key             Bytes
  ValueLength     Varint    (-1 = null)
  Value           Bytes
  HeadersCount    Varint
  Headers         [HeaderKeyLength Varint, HeaderKey String,
                   HeaderValueLength Varint, HeaderValue Bytes]

MAX_RECORD_OVERHEAD = 21 — "5 bytes length + 10 bytes timestamp + 5 bytes offset + 1 byte attributes", the worst case. In practice a record in a dense batch costs 6–8 bytes of framing.

Varints are ZigZag-encoded (Protobuf style), in ByteUtils. sizeOfVarint(-1) is precomputed as NULL_VARINT_SIZE_BYTES because null keys/values are extremely common (tombstones) and the size estimate is on the producer hot path.

A null value is a tombstone — semantically meaningful only in compacted topics, where it instructs the cleaner to remove the key. This is the one place Kafka's storage layer knows anything about record semantics.

3.4 Why the CRC is where it is

This is the single most interesting layout decision in the format, and the comment explains it:

The CRC covers the data from the attributes to the end of the batch [...] It is located after the magic byte, which means that clients must parse the magic byte before deciding how to interpret the bytes between the batch length and the magic byte. The partition leader epoch field is not included in the CRC computation to avoid the need to recompute the CRC when this field is assigned for every batch that is received by the broker.

Read that again. PartitionLeaderEpoch is written by the broker into a batch the producer CRC'd. If it were inside the CRC, every single produced batch would require a full CRC recomputation on the broker — on the hottest path in the system, over the full batch payload. By placing the field outside the checksummed range, the broker does a 4-byte in-place putInt and moves on.

The cost is that the magic byte sits after a field that isn't checksummed, so parsers must read magic first, then decide how to interpret preceding bytes. That's a genuinely awkward format. They took the awkwardness to save the CPU.

CRC-32C (Castagnoli) is used, not CRC-32, because it has a hardware instruction (SSE4.2 CRC32, ARMv8 CRC32C) and the JDK intrinsifies it. On a modern core that's ~1 byte/cycle vs. ~1 byte/8 cycles for a table-driven CRC-32.

Steal this: if you're designing a replicated log format, decide up front which fields are client-authored and which are server-authored, and put a checksum boundary between them.

3.5 The offset index

OffsetIndex is an mmapped array of 8-byte entries: (relativeOffset: int32, physicalPosition: int32).

Relative to the segment base offset — which is why a segment can only span Integer.MAX_VALUE offsets, and why AbstractIndex.toRelative returns OptionalInt.empty() on overflow, and why LogSegmentOffsetOverflowException exists.

It is sparse: an entry is appended only every log.index.interval.bytes (default 4096) of log written. So a lookup is: binary search the index for the largest entry ≤ target, then linear scan the log from that physical position. FileRecords.searchForOffsetFromPosition does the scan, and note how carefully it avoids lastOffset():

// The following logic is intentionally designed to minimize memory usage by avoiding
// unnecessary calls to lastOffset() for every batch.
// Instead, we use baseOffset() comparisons when possible, and only check lastOffset() when absolutely necessary.

baseOffset() is a field read from the batch header. lastOffset() is baseOffset + lastOffsetDelta — also a header read, but on a FileChannelRecordBatch it may force a header materialization. The loop is written to touch as few batch headers as possible.

The whole index for a 1 GB segment at default settings is 1 GB / 4 KB × 8 B = 2 MB. That's the entire index cost of Kafka's storage engine. Compare to an LSM tree's block index plus bloom filters plus manifest.

3.6 The warm-section binary search

This is the best-documented micro-optimization in the codebase and it's worth reproducing the argument, because it generalizes to any mmapped append-only index.

The setup: index files are mmapped, so reads go through the page cache. Appends always land at the end. Lookups (from in-sync followers and caught-up consumers) are also almost always near the end. So the pages near the end are hot and the rest are cold.

Now consider a standard binary search on a 13-page index, looking for something in page 12:

page number: |0|1|2|3|4|5|6|7|8|9|10|11|12 |
steps:       |1| | | | | |3| | |4|  |5 |2/6|

Pages 0, 6, 9, 11, 12 get touched every lookup, so they stay warm. Fine. Now the index grows to 14 pages:

page number: |0|1|2|3|4|5|6|7|8|9|10|11|12|13 |
steps:       |1| | | | | | |3| | | 4|5 | 6|2/7|

The probe set shifts to 0, 7, 10, 12, 13. Pages 7 and 10 have not been touched in a long time. The very next lookup after the index crosses a page boundary takes two cold page faults.

In our test, this can cause the at-least-once produce latency to jump to about 1 second from a few ms.

A one-second p99 spike, periodic, caused by binary search probe-set drift. The fix:

int firstHotEntry = Math.max(0, entries - 1 - warmEntries());
if (compareIndexEntry(parseEntry(idx, firstHotEntry), target, searchEntity) < 0) {
    return binarySearch(idx, target, searchEntity, searchResultType, firstHotEntry, entries - 1);
}
// ... else search [0, firstHotEntry]

warmEntries() = 8192 / entrySize() — i.e., the last 8 KB of the index. Two pages on a 4 KB-page machine. The comment justifies the constant on both sides:

• Not smaller: 8 KB of offset index ≈ 4 MB of log, so with default settings essentially all in-sync lookups land in the warm section.
• Not larger: a warm-section lookup touches exactly three entries — end, end-N, and (end*2-N)/2. With N = 8192 bytes and a ≥4 KB page, those three probes touch all ≤3 pages of the warm section on every lookup, so the entire warm section stays genuinely warm. Make N bigger and you'd have warm-section pages that aren't touched every time, which defeats the point.

The TODO at the end is honest about the remaining hole: low-QPS partitions have cold warm sections too, and the real fix would be a background thread that touches them.

Generalizable rule: when your data structure lives in the page cache and has a skewed access pattern, the algorithm's probe set stability matters as much as its asymptotic complexity. A O(log n) search whose probe set drifts is worse than an O(log n) search whose probe set is pinned.

3.7 Time index, txn index, epoch cache, producer snapshots

.timeindex — 12-byte entries (timestamp: int64, relativeOffset: int32), monotonically increasing in both. Backs ListOffsets by timestamp and time-based retention. Same warm-section search (warmEntries() = 8192/12 ≈ 682 entries).

.txnindex — TransactionIndex, a list of AbortedTxn records (producerId, firstOffset, lastOffset, lastStableOffset). Written on segment.updateTxnIndex(completedTxn, lastStableOffset) when an abort marker is appended. On a READ_COMMITTED fetch, the broker returns this list alongside the records and the consumer does the filtering. The broker never decompresses or filters user data for transactions — a deliberate choice to keep the fetch path zero-copy.

leader-epoch-checkpoint — a plain-text list of (epoch, startOffset). Maintained by LeaderEpochFileCache. This is the substrate of log reconciliation (§7.4). Note in UnifiedLog.append:

validRecords.batches().forEach(batch -> {
    if (batch.magic() >= RecordBatch.MAGIC_VALUE_V2) {
        assignEpochStartOffset(batch.partitionLeaderEpoch(), batch.baseOffset());
    } else {
        // In partial upgrade scenarios, we may get a temporary regression to the message format. In
        // order to ensure the safety of leader election, we clear the epoch cache so that we revert
        // to truncation by high watermark after the next leader election.
        if (leaderEpochCache.nonEmpty()) {
            logger.warn("Clearing leader epoch cache after unexpected append with message format v{}", batch.magic());
            leaderEpochCache.clearAndFlush();
        }
    }
});

The fallback path is the pre-KIP-101 "truncate to HW" behaviour, which is known to be able to lose data. It's preserved as a correctness-degrading-but-not-crashing escape hatch for mixed-format logs.

.snapshot — serialized ProducerStateManager state as of the segment base offset. On recovery, LogLoader finds the newest snapshot ≤ recovery point and replays forward, rather than scanning the whole log to rebuild producer sequence state. Retained for producer.id.expiration.ms worth of segments.

4. The write path, end to end

Let's trace a single ProduceRequest from the socket to fsync.

4.1 Acceptor / Processor / RequestChannel

core/src/main/scala/kafka/network/SocketServer.scala.

The threading model is a two-stage reactor:

 [Acceptor thread]  (1 per listener)
   ServerSocketChannel.accept()
   → round-robin assign to a Processor
   → processor.newConnections.offer(socketChannel)   [ArrayBlockingQueue, size 20]

 [Processor threads]  (num.network.threads per listener, default 3)
   loop:
     configureNewConnections()   // drain newConnections, register with selector
     processNewResponses()       // pull from responseQueue, register writes
     poll()                      // selector.poll(timeout)
     processCompletedReceives()  // → requestChannel.sendRequest(req)
     processCompletedSends()
     processDisconnected()
     closeExcessConnections()

 [RequestChannel]
   requestQueue: ArrayBlockingQueue[BaseRequest](queued.max.requests, default 500)

 [KafkaRequestHandler threads]  (num.io.threads, default 8)
   loop: requestChannel.receiveRequest() → KafkaApis.handle(request)

The interesting details:

Acceptor backpressure. newConnections is bounded at 20 per processor. If a processor is saturated, the acceptor blocks on newConnections.put() — which stops accepting. That's deliberate: better to leave connections in the kernel backlog than to accept them into a queue you can't serve.

Poll timeout. val pollTimeout = if (newConnections.isEmpty) 300 else 0 — if there are connections waiting to be configured, don't block in the selector at all.

Exception discipline. The processor loop catches Throwable at the top and keeps going:

We catch all the throwables here to prevent the processor thread from exiting. We do this because letting a processor exit might cause a bigger impact on the broker.

Losing a network thread silently degrades a third of your connection capacity with no crash. This is the right call for a broker, and the wrong call for most other software; know which one you're writing.

Channel muting. A channel is muted (removed from the read interest set) while its request is in flight. That's how Kafka enforces at most one in-flight request per connection server-side, which in turn is what makes max.in.flight.requests.per.connection=5 safe for idempotent producers — the broker processes a connection's requests strictly in order. handleChannelMuteEvent / tryUnmuteChannel implement it, and throttling piggybacks on the same mechanism (§13).

Memory pool. Selector holds a MemoryPool (queued.max.request.bytes). When memoryPool.availableMemory() < lowMemThreshold (10% of pool), the selector stops reading from all channels:

this.lowMemThreshold = (long) (0.1 * this.memoryPool.size());
...
if (!outOfMemory && memoryPool.availableMemory() < lowMemThreshold) { ... }

This is a global backpressure valve that pushes flow control down into TCP. Under sustained overload, receive windows close and producers block in send() rather than the broker OOMing. If you're building a broker, build this on day one — it is far harder to retrofit.

4.2 Request handler pool and callback re-entry

RequestChannel actually has two queues:

private val requestQueue = new ArrayBlockingQueue[BaseRequest](queueSize)
private val callbackQueue = new ArrayBlockingQueue[BaseRequest](queueSize)
...
def receiveRequest(timeout: Long): BaseRequest = {
  val callbackRequest = callbackQueue.poll()
  if (callbackRequest != null) callbackRequest
  else {
    val request = requestQueue.poll(timeout, TimeUnit.MILLISECONDS)
    request match {
      case _: WakeupRequest => callbackQueue.poll()
      case _ => request
    }
  }
}

The callback queue has strict priority. Its purpose: an operation that went async (a purgatory completion, a coordinator write, a remote-storage read) needs to finish on an I/O thread, and it must not queue behind 500 new produce requests. The WakeupRequest sentinel is pushed into requestQueue when a callback is enqueued, purely to unblock a handler that's parked in requestQueue.poll(timeout).

This is a nice pattern: a priority lane plus a wakeup token, rather than a priority queue (which would put a comparator on the hot path) or a separate thread pool (which would double the context switches).

4.3 ReplicaManager → Partition → UnifiedLog

KafkaApis.handleProduceRequest → ReplicaManager.appendRecords → ReplicaManager.appendToLocalLog → Partition.appendRecordsToLeader → UnifiedLog.appendAsLeader.

Partition.appendRecordsToLeader is where the min.insync.replicas check happens, before the append:

val minIsr = leaderLog.config.minInSyncReplicas.min(remoteReplicasMap.size + 1)

Note the .min(replicas + 1) — if you set min.insync.replicas=3 on a topic with RF=2, it's clamped rather than making the partition permanently unwritable. This clamp is newer than most people's mental model.

Partition holds the ISR machinery. Its partitionState field is a state machine:

CommittedPartitionState      — ISR is what the controller believes
PendingExpandIsr             — we've sent AlterPartition to add a replica, awaiting response
PendingShrinkIsr             — we've sent AlterPartition to remove replicas, awaiting response
OngoingReassignmentState     — adding/removing replicas mid-reassignment

The pending states are why maximalIsr exists (§7.3).

4.4 Anatomy of UnifiedLog.append

storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java:1115. The sequence, with the reasoning:

maybeFlushMetadataFile();                                          // topic id durable first
LogAppendInfo appendInfo = analyzeAndValidateRecords(...);         // CRC, sizes, offset monotonicity
if (appendInfo.validBytes() <= 0) return appendInfo;               // nothing to do
MemoryRecords trimmedRecords = trimInvalidBytes(records, appendInfo); // drop partial trailing batch
synchronized (lock) {                                              // ← the per-partition write lock
    // 1. offset assignment + validation + possible recompression
    // 2. leader epoch cache update
    // 3. maybeRoll
    // 4. producer state analysis (idempotence / txn)
    // 5. localLog.append  → FileRecords.append + index maybe-append
    // 6. updateHighWatermarkWithLogEndOffset
    // 7. txn index update, LSO advance
    // 8. flush if unflushedMessages >= flushInterval
}

Things worth calling out:

One lock per partition, and it covers real I/O. localLog.append writes to a FileChannel while holding the lock. It doesn't fsync (usually), so this is a page-cache write, but it is a syscall. Partition count is therefore your write concurrency — a topic with one partition has one writer, period. This is the fundamental reason Kafka's scaling unit is the partition and not the topic, and it's a property you inherit if you copy the design.

Trimming is silent. trimInvalidBytes drops a trailing partial batch without error. That's what makes it safe for a producer to send a truncated buffer, and it's also why a validBytes() <= 0 result returns a successful-looking LogAppendInfo with no offsets.

Two size checks, not one. There's a per-batch maxMessageSize check, and separately:

if (validRecords.sizeInBytes() > config().segmentSize()) {
    throw new RecordBatchTooLargeException(...);
}

You cannot append more than one segment's worth in a single call, because a segment must be able to hold at least one complete append.

Re-validation after recompression. If messageSizeMaybeChanged(), sizes are checked again — a broker recompressing from lz4 to zstd can change the batch size, and the check uses the original size for bytesRejectedRate to keep the metric comparable across the change.

Duplicate detection short-circuits the write entirely. If analyzeAndValidateProducerState returns a maybeDuplicate, the append info is filled in from the original batch's metadata and nothing is written. The producer gets back the offsets it got the first time. This is the whole of idempotent produce, and it's ~10 lines.

Ordering of LEO vs. txn index. The comment is explicit:

Append the records, and increment the local log end offset immediately after the append because a write to the transaction index below may fail, and we want to ensure that the offsets of future appends still grow monotonically. The resulting transaction index inconsistency will be cleaned up after the log directory is recovered.

They chose "monotonic offsets always, txn index may need repair" over "both or neither". Correct choice: offset monotonicity is an invariant every other subsystem depends on; the txn index is rebuildable.

4.5 LogValidator: offset assignment and recompression

LogValidator.validateMessagesAndAssignOffsets has three modes:

1. assignOffsetsNonCompressed — the fast path. Iterate batches, stamp baseOffset, lastOffsetDelta, partitionLeaderEpoch, timestamps, in place in the ByteBuffer. No copy, no CRC recompute (remember: leader epoch is outside the CRC).
2. validateMessagesAndAssignOffsetsCompressed — when source and target compression match and nothing forces a rewrite, it can still do in-place header stamping: it validates the inner records by decompressing into a BufferSupplier-provided buffer, then patches only the header.
3. buildRecordsAndAssignOffsets — the slow path. Full decompress → re-validate → recompress. Triggered by a compression type change (compression.type differs from producer's), a magic downgrade, or LogAppendTime with compressed input requiring timestamp rewrite.

BrokerCompressionType.targetCompression(config().compression, appendInfo.sourceCompression()) decides. compression.type=producer (the default) is what keeps you on paths 1–2. Setting a broker or topic compression.type that differs from what producers send silently puts every produce request on the recompression path. This is the single most common self-inflicted broker CPU problem in production.

The RequestLocal.bufferSupplier() threaded through here is a per-request-handler-thread buffer cache — decompression buffers are recycled per thread rather than allocated per batch. That's why append takes a RequestLocal at all.

4.6 Segment roll

maybeRoll(messagesSize, appendInfo) → LocalLog.roll. Roll triggers:

On roll: the old segment's indexes are trimmed to their actual size (resize() on the mmap — this is where AbstractIndex.resize and its remap write-lock earn their keep), a producer state snapshot is taken, and the new segment's files are created. Index trimming matters: an untrimmed index file is segment.index.bytes (10 MB) regardless of how full it is, so a broker with 50,000 segments would waste 500 GB of sparse-file page cache without it.

Roll is not free — it's file creation, mmap, mmap trim, and a producer snapshot, all under the partition lock. Very small segment.ms on a high-partition-count broker is a known way to make p99 produce latency ugly.

4.7 The fsync question

if (localLog.unflushedMessages() >= config().flushInterval) flush(false);

config().flushInterval is flush.messages, and its default is Long.MAX_VALUE. There is also flush.ms, defaulting to Long.MAX_VALUE.

Kafka does not fsync on the produce path. By default it never explicitly fsyncs a data segment at all except on roll and on clean shutdown. Durability comes from replication, not from the disk. acks=all means "in the page cache of min.insync.replicas machines", not "on the platters of any machine".

The reasoning, which I think is correct and which people still argue about:

• An fsync per produce request caps you at the device's sync IOPS. On NVMe that's survivable; on anything with a battery-backed cache it's fine; on EBS gp3 it is not.
• Correlated failure (a rack, an AZ, a bad kernel) defeats fsync anyway, and uncorrelated failure is exactly what replication handles.
• The page cache is a better cache than anything Kafka could build in the JVM: it's shared with the read path, it's not subject to GC, it survives broker restart, and sendfile can serve directly out of it.

The tail risk is real and named: a simultaneous power loss to min.insync.replicas machines loses acknowledged writes. If your deployment has correlated power domains, set flush.messages=1 and accept the IOPS bill, or put the replicas in different failure domains. There is no third option and the docs should say so more loudly.

Contrast with the KRaft metadata log, which does fsync — KafkaRaftLog flushes before responding to the leader, because Raft's safety proof requires durable votes and durable log entries. Kafka runs two different durability models in the same process, on purpose.

4.8 Idempotence: ProducerStateManager

Per-partition, per-producer-id state. ProducerStateEntry holds the last few BatchMetadata records (NUM_BATCHES_TO_RETAIN = 5):

BatchMetadata(lastSeq, lastOffset, offsetDelta, timestamp)

ProducerAppendInfo.append validates:

• Epoch fencing: incoming producerEpoch < currentEpoch → ProducerFencedException.
• Sequence continuity: expected lastSeq + 1. Gap → OutOfOrderSequenceException. This is fatal-ish, because the broker cannot know whether the missing batch was lost or is merely delayed.
• Duplicate: incoming (baseSequence, lastSequence) matches a retained BatchMetadata → return the cached offsets, write nothing.

Retaining 5 batches, not 1, is what makes max.in.flight.requests.per.connection=5 safe with enable.idempotence=true: a retry of batch n can arrive after n+1..n+4 have been accepted, and the broker still recognizes it as a duplicate rather than an out-of-order sequence.

Sequence numbers are int32 and wrap. Producer IDs are int64, allocated in blocks by the controller (ProducerIdControlManager / RPCProducerIdManager, ProducerIdsRecord in the metadata log) — a block per broker, so InitProducerId is usually a local operation.

State is expired after producer.id.expiration.ms (default 24h) of inactivity, which is also why a producer idle for longer than that can get UNKNOWN_PRODUCER_ID and must re-initialize.

The empty-batch-retention rule from the format comment ties in here:

if all of the records in a batch are removed during compaction, the broker may still retain an empty batch header in order to preserve the producer sequence information [...] retained only until either a new sequence number is written by the corresponding producer or the producerId is expired

Compaction must not destroy the sequence state that a rebuilt leader needs, or every producer would get a spurious OutOfOrderSequence after failover.

4.9 acks=all and the purgatory

ReplicaManager.appendRecords → local append succeeds → if acks == -1, maybeAddDelayedProduce builds a DelayedProduce and puts it in the produce purgatory keyed by TopicPartitionOperationKey per partition.

val delayedProduce = new DelayedProduce(timeoutMs, initialProduceStatus.asJava, delegate, responseCallback.asJava)
delayedProducePurgatory.tryCompleteElseWatch(delayedProduce, producerRequestKeys)

The completion trigger is Partition.maybeIncrementLeaderHW returning true — which happens when a follower's FetchRequest advances the leader's view of that follower's LEO. So the produce response latency for acks=all is:

produce_latency ≈ local_append + (time until every ISR follower's next fetch returns
                                  and its subsequent fetch reports the new LEO)

Note the subsequent fetch. The leader learns a follower has the data only when the follower's next fetch arrives with a higher fetchOffset. replica.fetch.wait.max.ms (default 500 ms) bounds how long a follower's fetch parks, but followers use fetch.min.bytes=1 so a fetch returns as soon as there's data — the practical latency is two network round trips plus the follower's local append.

DelayedProduce.tryComplete checks each partition's PartitionStatusValidator, which distinguishes: still-waiting, HW advanced past the required offset (success), leader changed (NotLeaderOrFollower), or ISR shrank below min.insync.replicas (NotEnoughReplicasAfterAppend — note the AfterAppend variant; the data is in the log and may still become committed, which is why this error is explicitly not safe to retry blindly under exactly-once).

5. The read path

5.1 Offset → file position

UnifiedLog.read(startOffset, maxLength, isolation, minOneMessage)
  → LocalLog.read
    → segments.floorEntry(startOffset)              // ConcurrentNavigableMap
    → segment.read(startOffset, maxSize, maxPosition, minOneMessage)
      → offsetIndex.lookup(startOffset)             // mmap binary search (warm section)
      → FileRecords.searchForOffsetFromPosition()   // linear scan from index hint
      → FileRecords.slice(position, length)         // no bytes read yet!

FileRecords.slice returns a new FileRecords sharing the same FileChannel with different start/end. No data has been read at this point. The FetchDataInfo handed back up the stack contains a lazy view over a file range.

maxPosition is the bound imposed by isolation level. For READ_UNCOMMITTED it's the HW's physical position; for READ_COMMITTED it's the LSO's. This is why LogOffsetMetadata carries (messageOffset, segmentBaseOffset, relativePositionInSegment) rather than just an offset — the fetch path needs the physical position of the HW without doing another index lookup.

minOneMessage handles the case where a single batch exceeds fetch.max.bytes: rather than return empty forever (a livelock), return the one oversized batch.

5.2 Zero-copy and where it breaks

FileRecords.writeTo:

public int writeTo(TransferableChannel destChannel, int offset, int length) throws IOException {
    long newSize = Math.min(channel.size(), end) - start;
    int oldSize = sizeInBytes();
    if (newSize < oldSize)
        throw new KafkaException("Size of FileRecords ... has been truncated during write");
    long position = start + offset;
    int count = Math.min(length, oldSize - offset);
    return (int) destChannel.transferFrom(channel, position, count);
}

transferFrom delegates to FileChannel.transferTo, which on Linux is sendfile(2). The path is page cache → socket buffer, entirely in the kernel. No user-space copy, no JVM heap allocation, no GC pressure proportional to fetch volume.

The truncation check is a real hazard: between building the FetchDataInfo and writing it, the log could be truncated (leader change) or deleted (retention). Failing loudly beats sending garbage.

TransferableChannel's javadoc names the sharp edge:

it will unwrap the destination channel, if possible, in order to benefit from zero copy. This is required because the fast path of transferTo is only executed if the destination buffer inherits from an internal JDK class.

If you wrap the socket channel in anything the JDK doesn't recognize, transferTo silently falls back to a read/write loop through a heap buffer. PlaintextTransportLayer.transferFrom unwraps carefully to stay on the fast path.

Zero-copy is lost when:

The TLS point is worth internalizing: Kafka's headline throughput numbers are plaintext numbers. If you must have TLS in the data path, budget for it, and consider whether sendfile-preserving alternatives (kTLS, offload NICs) are available to you.

5.3 Fetch sessions (KIP-227)

server/src/main/java/org/apache/kafka/server/FetchSession.java and FetchSessionCacheShard.java.

The problem: a follower fetching 5,000 partitions sends a FetchRequest naming all 5,000 partitions every ~100 ms, even though only a handful have new data. The request itself becomes the bottleneck.

The solution: the broker caches the session's partition set and last-fetched offsets. A follow-up "incremental" fetch sends only changes (added/removed partitions, changed offsets) plus a session ID and epoch. The response contains only partitions with data.

private int cachedSize = -1;   // last known size of this session; -1 = not in cache
private final int id;
private final boolean privileged;
private final ImplicitLinkedHashCollection<CachedPartition> partitionMap;
private volatile long lastUsedMs;
private volatile int epoch;

ImplicitLinkedHashCollection is a Kafka-specific data structure: a hash set where the link pointers live in the elements, so there's no per-entry Node object. With hundreds of thousands of cached partitions across sessions, that allocation saving is the difference between this being viable and not.

Eviction is two-tier and it's where the interesting policy lives. FetchSessionCacheShard keeps two TreeMaps:

private final TreeMap<EvictableKey, FetchSession> evictableByAll = new TreeMap<>();
private final TreeMap<EvictableKey, FetchSession> evictableByPrivileged = new TreeMap<>();

privileged = the session was created by a follower (replication), not a consumer. Followers can evict consumers; consumers cannot evict followers. The ordering key is (privileged, size, lastUsedMs) — bigger sessions are preferred for retention, since they save more. A session also becomes evictable-by-all once it's older than evictionMs:

if ((!session.privileged()) || (now - session.creationMs() > evictionMs))

Steal this: when you cache per-client state on a server, classify clients by criticality and make the eviction lattice explicit. "Replication traffic outranks consumer traffic" is a policy you want in the type system, not in a heuristic.

Failure mode to know: when the cache is full (max.incremental.fetch.session.cache.slots, default 1000), new sessions get INVALID_SESSION_ID and fall back to full fetches. This degrades silently into a throughput cliff. Watch NumIncrementalFetchSessions and IncrementalFetchSessionEvictionsPerSec.

5.4 DelayedFetch

If fetch.min.bytes isn't satisfied, the fetch goes into the fetch purgatory with fetch.max.wait.ms. DelayedFetch.tryComplete re-checks accumulated bytes and completes on:

• enough bytes accumulated
• the fetched partition's HW advanced (for follower fetches, LEO advanced)
• leader changed / partition moved / log truncated
• timeout

Completion is triggered from ReplicaManager.completeDelayedFetchRequests(topicPartitions), called after appends. So an acks=1 produce to a partition immediately unblocks consumers parked on that partition, without polling.

Note the interaction: produce completions and fetch completions are both purgatory operations watched on TopicPartitionOperationKey, and an append triggers checkAndComplete on both purgatories. One append can complete N delayed produces and M delayed fetches. The purgatory's estimatedTotalOperations / purgeInterval machinery exists to bound the cost of the watcher lists that build up from completed-but-not-yet-purged operations.

5.5 Read isolation: HW vs LSO

Two ceilings on what a consumer can see:

• High watermark (HW) — the highest offset replicated to all ISR members. READ_UNCOMMITTED consumers read up to HW.
• Last stable offset (LSO) — min(HW, firstUnstableOffset), where firstUnstableOffset is the first offset belonging to an open transaction. READ_COMMITTED consumers read up to LSO.

maybeIncrementFirstUnstableOffset() runs on every append. If a transaction opens at offset 1000 and stays open, LSO pins at 1000 even as HW runs to 10,000,000. A single hung transactional producer blocks every READ_COMMITTED consumer on that partition indefinitely — until transaction.max.timeout.ms fires and the coordinator aborts it. This is the number-one exactly-once operational surprise.

TransactionIndex supplies the abort list, and the consumer filters. In CompletedFetch/AbstractFetch, the consumer maintains a priority queue of aborted transactions by start offset and drops records whose producerId matches an open abort. Control batches (bit 5 of attributes) are consumed by the client and never surfaced to the application.

5.6 Tiered storage reads

RemoteLogManager (storage/src/main/java/org/apache/kafka/server/log/remote/storage/). Two SPIs:

• RemoteStorageManager — bytes. copyLogSegmentData, fetchLogSegment, fetchIndex, deleteLogSegmentData.
• RemoteLogMetadataManager — metadata. Default implementation (TopicBasedRemoteLogMetadataManager) stores it in an internal Kafka topic, __remote_log_metadata. Kafka storing Kafka's metadata in Kafka.

Three task pools:

private final ConcurrentHashMap<TopicIdPartition, RLMTaskWithFuture> leaderCopyRLMTasks;
private final ConcurrentHashMap<TopicIdPartition, RLMTaskWithFuture> leaderExpirationRLMTasks;
private final ConcurrentHashMap<TopicIdPartition, RLMTaskWithFuture> followerRLMTasks;

Copy and expiration are separate pools — that split was added because slow expiration (S3 DELETE throttling) was starving the copy path and causing local disks to fill.

Only segments below the high watermark and fully rolled are copied. Local retention (local.retention.ms / .bytes) then deletes them locally; total retention governs remote.

The read path. ReplicaManager.readFromLog gets an OffsetOutOfRangeException for the local log when the requested offset is below logStartOffset but above the remote start, and builds a RemoteStorageFetchInfo. Then:

remoteFetchTask = remoteLogManager.get.asyncRead(remoteFetchInfo, (result: RemoteLogReadResult) => {
  remoteFetchResult.complete(result)
  ...
})

and a DelayedRemoteFetch goes into the purgatory. The I/O thread is not blocked on S3. A separate RemoteStorageThreadPool does the fetch; the purgatory completes when the future does. Without this, one slow object-store read would consume an num.io.threads slot for its full latency.

RemoteIndexCache caches remote offset/time/txn indexes on local disk (LRU, sized by remote.log.index.file.cache.total.size.bytes) so a remote fetch doesn't re-download the index.

The honest performance picture: remote reads are ~100 ms p50 against S3 vs. ~1 ms from page cache, and they are not zero-copy. Tiered storage is for retention economics and for backfill readers, not for latency-sensitive consumers. Design your consumer groups so that lagging consumers are the ones hitting remote.

6. Hierarchical timing wheels and the purgatory

server-common/src/main/java/org/apache/kafka/server/util/timer/TimingWheel.java is one of the cleanest pieces of code in the tree, and the comment is a small paper.

The problem: hundreds of thousands of concurrent timeouts (every delayed produce, delayed fetch, delayed join, heartbeat expiry), and the overwhelming majority are cancelled before firing. A DelayQueue or ScheduledThreadPoolExecutor gives you O(log n) insert and O(log n) delete. When 99% of your operations are insert-then-delete, that's the wrong shape.

A simple timing wheel is a circular array of n buckets each covering u time units:

A timing wheel has O(1) cost for insert/delete (start-timer/stop-timer) whereas priority queue based timers [...] have O(log n) insert/delete cost.

The drawback is bounded range: n * u. The fix is hierarchy — level k+1 has resolution n times coarser:

level    buckets
1        [c,c]   [c+1,c+1]  [c+2,c+2]
2        [c,c+2] [c+3,c+5]  [c+6,c+8]
3        [c,c+8] [c+9,c+17] [c+18,c+26]

Overflow delegates upward; when a higher bucket expires, its tasks are reinserted and cascade down to finer wheels. Insert is O(m) where m = number of levels (tiny), delete is O(1).

The O(1) delete is the crux, and it comes from TimerTaskList being a doubly-linked list where the TimerTaskEntry is held by the TimerTask itself. Cancelling is entry.remove() — unlink two pointers. No search.

Note the overlap the comment calls out:

bucket [c,c+2] in level 2 won't receive any task since that range is already covered in level 1. [...] This is a bit wasteful, but simplifies the implementation.

A maintainer accepting a small constant-factor waste for a large simplicity win, and documenting it. More code should do this.

How the driving thread works. The wheels themselves don't have a thread. SystemTimer holds a DelayQueue<TimerTaskList> of buckets, not tasks. So the JDK's DelayQueue — with its O(log n) ops — is used over the number of non-empty buckets, which is small and bounded, while the number of tasks is unbounded. This is the trick: use the expensive structure over the cheap dimension.

SystemTimer.advanceClock polls the bucket queue, and an "expiring timer" thread (ExpirationReaper per purgatory) drives it.

The purgatory (DelayedOperationPurgatory) layers a watcher-list index on top:

public <K extends DelayedOperationKey> boolean tryCompleteElseWatch(T operation, List<K> watchKeys) {
    if (operation.safeTryCompleteOrElse(() -> {
        watchKeys.forEach(key -> { if (!operation.isCompleted()) watchForOperation(key, operation); });
        if (!watchKeys.isEmpty()) estimatedTotalOperations.incrementAndGet();
    })) return true;
    if (!operation.isCompleted()) {
        if (timerEnabled) timeoutTimer.add(operation);
        if (operation.isCompleted()) operation.cancel();
    }
    return false;
}

The double-check (try, then watch, then try again under the operation's lock) is the standard "register before re-checking" pattern that avoids missing a concurrent trigger. The comment then enumerates a real deadlock they hit:

1. thread_a holds readlock of stateLock from TransactionStateManager
2. thread_a is executing tryCompleteElseWatch()
3. thread_a adds op to watch list
4. thread_b requires writelock of stateLock (blocked by thread_a)
5. thread_c calls checkAndComplete() and holds lock of op
6. thread_c is waiting readlock of stateLock (blocked by thread_b)
7. thread_a is waiting lock of op (blocked by thread_c)

...and then admits the current fix doesn't eliminate the class, only this instance, and states the convention that actually keeps it safe:

we recommend DelayedOperationPurgatory.checkAndComplete() be called without holding any exclusive lock.

An invariant maintained by convention and comment rather than by construction. Worth knowing if you touch this code.

Reaper / purge. Completed operations stay in watcher lists until purged. purgeInterval (producer.purgatory.purge.interval.requests, default 1000) controls when a scan reclaims them. estimatedTotalOperations vs. actual watched() is the signal. Symptom of getting this wrong: purgatory size metrics climbing without bound while actual pending work is small.

7. Replication

7.1 Pull replication, and why

Followers fetch. There is no push. ReplicaFetcherThread extends AbstractFetcherThread issues FetchRequest to the leader with replicaId set, exactly like a consumer but with privileges (reads above the HW, gets a privileged fetch session, gets OffsetForLeaderEpoch).

Consequences, all of which you inherit if you copy this:

• The leader has no per-follower send queue. No unbounded buffering, no head-of-line blocking from a slow follower. A slow follower just falls behind and eventually leaves the ISR.
• The replication path is the consumer path. Same FetchRequest, same purgatory, same sendfile. One code path to optimize. Fewer bugs.
• The HW lags by one round trip (§7.2). This is the price.
• Adding a replica is trivial — it's just a new fetcher. Compare to push-based systems where the leader must track and manage each replica's stream.

AbstractFetcherManager partitions the work: num.replica.fetchers threads per source broker, partitions hashed across them. A broker with 4,000 partitions replicating from 10 leaders and num.replica.fetchers=4 has 40 fetcher threads, each multiplexing ~100 partitions into one incremental fetch session.

7.2 The high watermark, and its one-round-trip lag

Partition.maybeIncrementLeaderHW:

private def maybeIncrementLeaderHW(leaderLog: UnifiedLog, currentTimeMs: Long = time.milliseconds): Boolean = {
  if (isUnderMinIsr) {
    trace(s"Not increasing HWM because partition is under min ISR(ISR=${partitionState.isr})")
    return false
  }
  val leaderLogEndOffset = leaderLog.logEndOffsetMetadata
  var newHighWatermark = leaderLogEndOffset
  remoteReplicasMap.forEach { (_, replica) =>
    val replicaState = replica.stateSnapshot
    def shouldWaitForReplicaToJoinIsr: Boolean =
      replicaState.isCaughtUp(leaderLogEndOffset.messageOffset, currentTimeMs, replicaLagTimeMaxMs) &&
      isReplicaIsrEligible(replica.brokerId)
    if (replicaState.logEndOffsetMetadata.messageOffset < newHighWatermark.messageOffset &&
        (partitionState.maximalIsr.contains(replica.brokerId) || shouldWaitForReplicaToJoinIsr)) {
      newHighWatermark = replicaState.logEndOffsetMetadata
    }
  }
  leaderLog.maybeIncrementHighWatermark(newHighWatermark) match { ... }
}

Three things:

1. isUnderMinIsr short-circuits. If the ISR is below min.insync.replicas, the HW does not advance at all — even for data that is replicated to everyone currently in the ISR. This is what makes acks=all + min.insync.replicas=2 actually mean something rather than degrading to acks=1 when the ISR shrinks to one.
2. shouldWaitForReplicaToJoinIsr holds the HW back for a replica that's caught up but not yet in the ISR. Without it, the HW could advance past a point, then the catching-up replica joins the ISR, and you'd have an ISR member behind the HW — which breaks the leader-election invariant.
3. The comment marker "using the maximal" — see §7.3.

The lag: the leader learns follower LEO k from a fetch at fetchOffset = k. It then advances HW to k. Followers learn the new HW from the next fetch response. So a follower's HW trails the leader's by one fetch round trip. This is why:

• On leader failover, the new leader truncates to its HW, which may be behind — and this is why KIP-101 leader epochs were needed to avoid data loss/divergence during that truncation.
• log.flush and consumer visibility are both bounded by this, and no amount of tuning removes it.

If you're designing this: a push-based system (Raft AppendEntries) can commit in one round trip because the leader knows immediately that a follower accepted. Kafka trades one RTT of commit latency for a vastly simpler and more uniform replication path. Whether that's the right trade depends on whether your commit latency budget is ~5 ms or ~500 µs.

7.3 ISR: expand, shrink, maximal ISR

Expand (maybeExpandIsr, triggered by a follower fetch):

private def isFollowerInSync(followerReplica: Replica): Boolean = {
  ...
  followerEndOffset >= leaderLog.highWatermark && leaderEpochStartOffsetOpt.exists(followerEndOffset >= _)
}

Two conditions, and the second is the subtle one: the follower must be caught up to the current leader epoch's start offset, not just the HW. Otherwise a follower could join the ISR while still holding records from a previous, diverged epoch.

Shrink (maybeShrinkIsr, from a scheduled task): a follower is out of sync if it hasn't fetched up to the leader's LEO within replica.lag.time.max.ms (default 30s). Note this is purely time-based — the old replica.lag.max.messages was removed years ago because a message-count threshold produces ISR flapping under bursty load.

Both go through the controller. The leader sends AlterPartitionRequest; the controller validates (leader epoch, partition epoch) and writes a PartitionChangeRecord to the metadata log. Only when that's committed and propagated back does the leader's partitionState become CommittedPartitionState again.

maximalIsr exists because of the window in between. When PendingExpandIsr is active:

isr        = {1, 2}          # what the controller has committed
maximalIsr = {1, 2, 3}       # what we've proposed

For HW advancement, the leader uses maximalIsr — pessimistic, waits for replica 3 too. For min.insync.replicas checks it uses the committed isr — pessimistic in the other direction. Both choices are the conservative one, and that's the point: during the uncertainty window, assume whichever set makes you safer for the property in question.

PendingShrinkIsr is symmetric: HW advancement still waits for the replicas you're trying to remove until the controller confirms the removal. If it didn't, you could advance the HW past data that the about-to-be-removed replica has but the survivors don't — and then fail over to a survivor.

7.4 Leader epochs and log reconciliation

This is KIP-101, and it's the mechanism that lets Kafka use pull replication without divergence.

leader-epoch-checkpoint maps epoch → start offset. On becoming a follower, AbstractFetcherThread puts the partition in Truncating state and runs:

1. For each truncating partition, take its latest (epoch, LEO).
2. Send OffsetForLeaderEpochRequest(epoch) to the leader.
   (Or, on modern versions, the leader piggybacks `divergingEpoch` on the FetchResponse —
    see truncateOnFetchResponse.)
3. Leader responds with the end offset of that epoch in ITS log.
4. Follower truncates to min(leaderEndOffsetForEpoch, ownLEO).
5. Repeat with the next-lower epoch if needed.

The key insight: the epoch boundary tells you exactly where two logs can diverge. If the leader says "epoch 5 ended at offset 1000" and you have records at offset 1200 in epoch 5, those 200 records were never committed and must go. You don't need to compare record contents; you don't need to truncate blindly to the HW.

truncateToHighWatermark still exists as the fallback for logs with no epoch information (mixed format, pre-KIP-101 data). It's marked as the unsafe path.

The modern optimization (truncateOnFetchResponse) piggybacks divergingEpoch on the fetch response, so the common case needs no extra RPC at all:

.setLeaderEpoch(partitionData.divergingEpoch.epoch)
.setEndOffset(partitionData.divergingEpoch.endOffset)

Design lesson: if your replication protocol truncates, you need a ratchet that identifies the divergence point without content comparison. Raft uses (prevLogIndex, prevLogTerm) on every AppendEntries. Kafka uses a persisted epoch→offset map queried on demand. Both work; Kafka's version has the advantage of being queryable in bulk for thousands of partitions in one RPC.

7.5 Eligible Leader Replicas (KIP-966)

The pre-ELR world had exactly two options when all ISR members are down:

• unclean.leader.election.enable=false: partition unavailable until an ISR member returns.
• unclean.leader.election.enable=true: elect anyone, silently lose data.

ELR adds a third. PartitionRegistration gains elr[] and lastKnownElr[]. When a replica leaves the ISR because its broker shut down or was fenced — not because it fell behind — it goes into the ELR. An ELR member is known to have had all committed data at the moment it left. So:

if (isValidNewLeader(preferredReplica)) return new ElectionResult(preferredReplica, false);
if (isValidNewLeader(partition.leader))  return ...
// then: any ISR member
// then: any ELR member   ← new, and still a CLEAN election
// then (only if unclean enabled): any live replica

PartitionChangeBuilder maintains it, gated on eligibleLeaderReplicasEnabled (a MetadataVersion feature, IBP_4_0_IV1+ with elr.version). BrokersToElrs is the controller-side reverse index (broker → partitions where it's in the ELR), so that when a broker comes back the controller can immediately find the partitions it can rescue.

lastKnownElr handles the fully-cold-cluster case: if ISR and ELR are both empty, the controller can still elect the last known ELR member — but only if there's exactly one, and only under useLastKnownLeaderInBalancedRecovery:

if (partition.lastKnownElr.length != 1) {
    log.trace("Try to elect last known leader for {}-{} but lastKnownElr does not only have 1 member...");

More than one candidate means you can't tell which had more data, so it refuses.

This is a genuinely good piece of distributed systems design and it's underappreciated. The insight is that "left the ISR" conflates two very different events — fell behind (may be missing data) and went away (had everything up to the moment it went away) — and that distinguishing them recovers availability with no safety loss.

8. KRaft

ZooKeeper is gone — removed entirely in 4.0. KafkaRaftServer starts a ControllerServer, a BrokerServer, or both (combined mode), sharing a SharedServer that owns the KafkaRaftClient.

8.1 How it differs from textbook Raft

The class comment is explicit and worth quoting in full:

This class implements a Kafkaesque version of the Raft protocol. Leader election is more or less pure Raft, but replication is driven by replica fetching and we use Kafka's log reconciliation protocol to truncate the log to a common point following each leader election.

The API set:

The comment explains why BeginQuorumEpoch exists at all:

This is not needed in usual Raft because the leader can use an empty data push to achieve the same purpose. The Kafka Raft implementation, however, is driven by fetch requests from followers, so there must be a way to find the new leader after an election has completed.

That's the fundamental tension of pull-based consensus: the followers don't know who to pull from until someone tells them. BeginQuorumEpoch is retried indefinitely per voter until acknowledged.

Similarly, truncation detection is piggybacked on Fetch rather than being a separate protocol state, matching the partition replication design.

Observers vs. voters. Brokers are observers of __cluster_metadata: they fetch, they never vote, they never become leader. Controllers are voters. This is the same replicaId-based distinction as partition replication, and it means adding a broker doesn't touch the quorum.

MAX_BATCH_SIZE_BYTES = 8 * 1024 * 1024, MAX_FETCH_WAIT_MS = 500, MAX_NUMBER_OF_BATCHES = 10.

8.2 The quorum state machine, including Prospective

QuorumState.java's comment is the authoritative state diagram:

Resigned →     Unattached (higher epoch, or election timeout)
               Follower   (discovered leader with larger epoch)

Unattached →   Unattached (higher epoch, or after giving a binding vote)
               Prospective (election timeout)
               Follower   (discovered leader, equal or larger epoch)

Prospective →  Unattached  (higher epoch; or no last known leader and lost/timed out)
               Candidate   (majority of PreVotes granted)
               Follower    (larger epoch; or had a last known leader and lost/timed out)

Candidate →    Unattached  (higher epoch)
               Prospective (election timeout or loss)
               Leader      (majority of votes)

Leader →       Unattached  (higher epoch)
               Resigned    (graceful shutdown)
               Follower    (larger epoch)

Follower →     Unattached  (higher epoch)
               Prospective (fetch timeout)
               Follower    (larger epoch)

Prospective is the pre-vote state (KIP-996) and it's the most important addition to KRaft's election protocol. ProspectiveState's javadoc:

1. Once started, it will send prevote requests and keep record of the received vote responses
2. If it receives a message denoting a leader with a higher epoch, it will transition to follower
3. If majority votes granted, it will transition to candidate state
4. If majority votes rejected or election times out, it will transition to unattached or follower depending on if it knows the leader id and endpoints or not

The problem pre-vote solves: a partitioned node's election timer fires repeatedly. Each time, it bumps its epoch and campaigns. When the partition heals, its high epoch forces the healthy leader to step down — even though the partitioned node was never a viable leader and has a stale log. In a metadata quorum, that means controller churn and a metadata write stall for no reason.

Pre-vote: campaign without bumping the epoch. Ask "would you vote for me?" Voters answer based on log up-to-dateness and on whether they currently have a healthy leader. Only on a majority of yeses do you bump the epoch and run a real election. canGrantVote(replicaKey, isLogUpToDate, isPreVote) carries the flag through.

Note also the transition rule that a prospective can re-enter Unattached in the same epoch:

if (epoch < currentEpoch || (epoch == currentEpoch && !isProspective())) {
    throw new IllegalStateException(...);
}

Same-epoch backward transitions are normally illegal in Raft; they're legal here specifically because pre-vote didn't bump the epoch, so no state was published.

Complementary is checkQuorum on the leader side (LeaderState):

int majority = (voterStates.size() / 2) + 1;
if (!voterStates.containsKey(localReplicaKey)) majority = majority - 1;
if (fetchedVoters.size() >= majority) { ...reset timer... }

Did not receive fetch request from the majority of the voters within {}ms.

A leader that stops hearing from a majority resigns on its own, rather than waiting to be displaced. That closes the other half of the partitioned-leader problem.

8.3 Leader HW computation

LeaderState.updateHighWatermark:

// Find the largest offset which is replicated to a majority of replicas (the leader counts)
Optional<LogOffsetMetadata> highWatermarkUpdateOpt = followersByDescendingFetchOffset.get(indexOfHw).endOffset;

Sort replicas by fetch offset descending, take the element at the majority index. Standard Raft commit-index computation. Two guards:

if (highWatermarkUpdateOffset > epochStartOffset) { ... }

Raft's rule that a leader may not commit an entry from a previous term by counting replicas — it must commit something from its own term first. epochStartOffset is the enforcement point. And:

} else if (highWatermarkUpdateOffset < currentHighWatermarkMetadata.offset()) {
    log.info("The latest computed high watermark {} is smaller than the current value {}, ...");

HW is monotonic; a decrease means a bug and is logged loudly rather than applied.

The Optional<LogOffsetMetadata> (vs. a bare long) is again the physical-position optimization: the FetchSnapshot/Fetch responder needs the byte position of the HW.

8.4 Snapshots

__cluster_metadata cannot be compacted with the normal log cleaner — the records are deltas (PartitionChangeRecord), not key-value overwrites. So KRaft uses snapshots: a full serialized MetadataImage at (offset, epoch), written as 00000000000000012345-0000000042.checkpoint.

RecordsSnapshotWriter / RecordsSnapshotReader use the same FileRecords machinery as segments, so FetchSnapshot can sendfile too. UnalignedFileRecords exists because snapshot content isn't offset-aligned — the position→offset relationship that FileRecords assumes doesn't hold, so a distinct type prevents accidental misuse.

NotifyingRawSnapshotWriter fires a callback on completion so the log can advance its start offset and delete the now-redundant prefix.

Generation is triggered by metadata.log.max.record.bytes.between.snapshots (default 20 MB) or metadata.log.max.snapshot.interval.ms (default 1h).

The bootstrap snapshot (BOOTSTRAP_SNAPSHOT_ID, offset -1) is how kafka-storage format seeds initial FeatureLevelRecords before any log exists — the bootstrap.checkpoint file.

KRaftControlRecordStateMachine tracks control records embedded in the log (voter set changes, KRaftVersionRecord), and VoterSetHistory / TreeMapLogHistory keep the voter set as of any offset — necessary because a Vote request must be evaluated against the voter set that was current at the relevant offset, not the latest one. Dynamic quorum membership (KIP-853) is what requires this; AddVoterHandler / RemoveVoterHandler / UpdateVoterHandler implement the one-at-a-time membership change protocol.

8.5 QuorumController: a single-threaded event loop

metadata/src/main/java/org/apache/kafka/controller/QuorumController.java. This is the piece that most repays study if you're building a control plane.

Everything is an event on one queue. KafkaEventQueue, one thread. Every mutation is a ControllerWriteEvent:

class ControllerWriteEvent<T> implements EventQueue.Event, DeferredEvent {
    // run():
    //   1. verify still active controller (epoch check)
    //   2. call op.generateRecordsAndResult()  → ControllerResult<T>(records, response)
    //   3. raftClient.scheduleAtomicAppend(records)
    //   4. deferredEventQueue.add(resultAndOffset.offset(), this)   ← does NOT complete the future
    // and later, when the offset is committed:
    //   deferredEventQueue.completeUpTo(offsetControl.lastStableOffset());
}

The design properties this buys:

1. No locks anywhere in the control managers. ReplicationControlManager, ClusterControlManager, ConfigurationControlManager, FeatureControlManager are all single-threaded-by-construction. Read their code — there is not a synchronized or a ConcurrentHashMap in the state. That's an enormous simplification for logic this intricate.
2. Deterministic replay. The controller's state is exactly fold(replay, records). A standby controller applying the same records reaches the same state. This is testable — and test-common/ has harnesses that do exactly that.
3. Responses are deferred until commit. The client's CreateTopicsResponse doesn't arrive until the TopicRecord is committed to a majority of the quorum. No dirty reads of uncommitted controller state ever escape.
4. Snapshottable state. The managers' state lives in TimelineHashMap / TimelineHashSet (server-common/.../timeline/) — persistent data structures with epoch-tagged versions, so the controller can serve reads at a past offset and can revert state if a write is not committed (leadership loss). This is the mechanism that makes "generate records, respond later" safe: if the append fails, the in-memory state is rolled back to the last committed offset.

ControllerOperationFlag carries per-operation policy (e.g. DOES_NOT_UPDATE_QUEUE_TIME for periodic tasks, RUNS_IN_PREMIGRATION).

EventPerformanceMonitor logs slow events. On a controller managing a million partitions, a single event that takes 500 ms stalls all metadata mutation. The monitor exists because that's the dominant failure mode of a single-threaded design and you need it visible.

PeriodicTaskControlManager schedules background work (ELR cleanup, expired-token removal, partition-leader balancing) as events on the same queue, so background work is serialized with foreground work and can't race it.

ActivationRecordsGenerator produces the records written on becoming active controller — a tricky bit of code that handles bootstrap, MetadataVersion upgrades detected at activation, and migration-era leftovers.

Steal all of this. The pattern — single-threaded deterministic state machine, replicated log for durability, deferred completion at commit, persistent data structures for rollback — is the single highest-leverage design in the Kafka codebase. It converts a nightmarishly concurrent problem (cluster metadata mutation) into sequential code that reads like a textbook.

The cost is honest too: one core, and a stall anywhere stalls everything. Kafka's answer is aggressive monitoring plus keeping the per-event work small. If your control plane has genuinely CPU-heavy operations, you need a different answer (batching, or sharding the state machine).

8.6 MetadataImage / MetadataDelta / MetadataLoader

MetadataImage   — immutable snapshot of all cluster metadata at an offset
MetadataDelta   — a mutable builder holding changes since an image
MetadataLoader  — reads the raft log, builds deltas, publishes images
MetadataPublisher — consumers of images/deltas (ReplicaManager, coordinators, quota managers, ...)

public void replay(ApiMessage record) { /* dispatch by record type */ }
public void replay(TopicRecord record)           { getOrCreateTopicsDelta().replay(record); }
public void replay(PartitionChangeRecord record) { getOrCreateTopicsDelta().replay(record); }
public void replay(FeatureLevelRecord record)    { ... }
public void replay(ClearElrRecord record)        { ... }
...

Lazy sub-delta creation (if (topicsDelta == null) topicsDelta = new TopicsDelta(image.topics())) means a batch that only touches configs allocates nothing in the topics dimension. With a million partitions, that's the difference between a metadata batch costing microseconds and costing a full-image copy.

MetadataBatchLoader accumulates records until a batch boundary before publishing — publishers see transactionally-consistent images, never a half-applied batch.

TopicsDelta.localChanges(brokerId) produces a LocalReplicaChanges — exactly the set of leader/follower/delete transitions this broker must perform. ReplicaManager.applyDelta consumes it. This replaces the ZooKeeper-era LeaderAndIsrRequest broadcast entirely: instead of the controller computing and sending per-broker instructions, each broker derives its own instructions from the shared log. That's a fundamental architectural improvement — the controller no longer needs to know what each broker has already processed.

Failure mode this eliminated: in the ZK world, a broker that missed a LeaderAndIsrRequest had stale state and the controller had to detect and resend. In KRaft, a broker that's behind is simply behind in the log, and catches up by fetching. Divergence is impossible by construction.

8.7 Feature flags and MetadataVersion

server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:

IBP_4_0_IV1(23, "4.0", "IV1", true),
IBP_4_0_IV2(24, "4.0", "IV2", false),
IBP_4_0_IV3(25, "4.0", "IV3", false),
IBP_4_1_IV0(26, "4.1", "IV0", false),
IBP_4_1_IV1(27, "4.1", "IV1", false),
IBP_4_2_IV0(28, "4.2", "IV0", false),
IBP_4_2_IV1(29, "4.2", "IV1", false),
IBP_4_3_IV0(30, "4.3", "IV0", true),
IBP_4_4_IV0(31, "4.4", "IV0", false),   // dead-letter queue for share groups (KIP-1191)
IBP_4_4_IV1(32, "4.4", "IV1", true),
IBP_4_4_IV2(33, "4.4", "IV2", true);

public static final MetadataVersion LATEST_PRODUCTION = IBP_4_3_IV0;

The boolean is didMetadataChange — whether the version introduced new metadata record types or versions, which determines whether a downgrade is possible. Versions above LATEST_PRODUCTION are "testing" and require unstable.metadata.versions.enable.

Capability checks read as predicates on the version:

public boolean isElrSupported()        { return this.isAtLeast(IBP_4_0_IV1); }
public boolean isShareGroupDLQSupported() { return this.isAtLeast(IBP_4_4_IV1); }

Beyond metadata.version, there are independent features (kraft.version, transaction.version, group.version, eligible.leader.replicas.version, share.version, streams.version) each with their own level, stored as FeatureLevelRecords and managed by FeatureControlManager. QuorumFeatures and ClusterFeatureSupportDescriber compute the max level the whole cluster supports, so a feature can't be enabled above what the oldest node understands.

This is how you do rolling upgrades in a replicated-log system: the log is the source of truth for what's enabled, the enabling record is itself in the log, and every node's behaviour is a pure function of the log prefix it has applied. There's no "wait for everyone to restart" step and no distributed agreement problem separate from the log.

9. Coordinators

9.1 The coordinator runtime

coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/ — CoordinatorRuntime is a generic framework, and the group, share, and (in progress) transaction coordinators are all instances of it. It is essentially QuorumController's design applied to a partitioned state machine.

CoordinatorRuntime<S extends CoordinatorShard<U>, U>
  ├─ one CoordinatorShard per partition of the backing internal topic
  ├─ CoordinatorEventProcessor — striped executor, events for a shard are serialized
  ├─ coordinator writes produce records → appended to __consumer_offsets / __share_group_state
  └─ responses deferred until the append is committed (HW advances past it)

The properties mirror the controller: single-threaded per shard (so GroupMetadataManager has no locks), deterministic replay from the log, and responses that don't escape before durability.

The difference is sharding — 50 partitions of __consumer_offsets by default means 50 independent state machines and 50 threads' worth of parallelism, which is what lets one broker coordinate tens of thousands of groups.

CoordinatorLoader replays a partition's records into a shard on becoming leader for it. SnapshotRegistry + timeline collections again provide the rollback-on-uncommitted-write property.

9.2 Group coordinator: KIP-848

The old protocol (JoinGroup/SyncGroup, "classic") had a structural problem: the assignment was computed by an elected group leader — a client. That produced:

• Stop-the-world rebalances. Every member revokes everything, rejoins, waits for the leader to compute, gets a new assignment. A 500-member group with a 100 ms assignor takes seconds of total unavailability.
• Client-side assignor code that the broker couldn't validate or version.
• session.timeout.ms / max.poll.interval.ms coupling that made a slow processing loop look like a dead member.

KIP-848 moves assignment server-side and makes reconciliation incremental and per-member. The protocol is a single RPC:

ConsumerGroupHeartbeat(groupId, memberId, memberEpoch, subscribedTopicNames|regex,
                       rebalanceTimeoutMs, topicPartitions[ownedPartitions])
  → ConsumerGroupHeartbeatResponse(memberEpoch, heartbeatIntervalMs, assignment)

That's it. No join, no sync, no leader. ConsumerGroupHeartbeatRequest carries the member's owned partitions; the response carries its target. The state machine in between does the rest.

The coordinator maintains:

• Group epoch — bumped on any subscription/membership/metadata change.
• Target assignment — computed by a server-side assignor when the group epoch changes, stored as ConsumerGroupTargetAssignmentMemberRecords. Has its own epoch.
• Per-member current assignment — reconciled toward the target, one member at a time, independently.

ConsumerGroupHeartbeat also carries subscribedTopicRegex — server-side regex subscription (ResolvedRegularExpression records), which removes the old requirement that every member have metadata for every topic in order to evaluate the pattern consistently.

9.3 The reconciliation state machine

group-coordinator/.../modern/consumer/CurrentAssignmentBuilder.java:

The CurrentAssignmentBuilder class encapsulates the reconciliation engine of the consumer group protocol. Given the current state of a member and a desired or target assignment state, the state machine takes the necessary steps to converge them.

Three member states (modern/MemberState.java):

STABLE               // fully reconciled with the target assignment
UNREVOKED_PARTITIONS // must revoke some partitions before advancing epoch
UNRELEASED_PARTITIONS// advanced epoch, waiting on partitions not yet revoked by previous owners
UNKNOWN              // forward-compat: a state this version doesn't know

build():

case STABLE:
    if (member.memberEpoch() != targetAssignmentEpoch) return computeNextAssignment(...);
    else if (hasSubscriptionChanged) return updateCurrentAssignment(...);
    else return member;

case UNREVOKED_PARTITIONS:
    // revocation is confirmed by ABSENCE from the heartbeat's owned-partitions list
    if (ownsRevokedPartitions(member.partitionsPendingRevocation())) {
        if (hasSubscriptionChanged) return updateCurrentAssignment(...);
        else return member;                      // still waiting — no progress
    }
    return computeNextAssignment(...);

case UNRELEASED_PARTITIONS:
    return computeNextAssignment(...);           // pick up partitions as they free up

case UNKNOWN:
    if (ownedTopicPartitions == null || !ownedTopicPartitions.isEmpty())
        throw new FencedMemberEpochException("The consumer group member is in a unknown state. "
            + "The member must abandon all its partitions and rejoin.");
    return computeNextAssignment(targetAssignmentEpoch, member.assignedPartitions());

Three details worth extracting:

Revocation is proven by omission. The coordinator never asks "did you revoke?" — it observes that the partition stopped appearing in ownedTopicPartitions. That's a level-triggered protocol, not edge-triggered: a lost heartbeat costs you a round trip, not correctness. Enormously more robust than an ack-based revoke.

currentPartitionEpoch is the safety interlock.

/**
 * A function which returns the current epoch of a topic-partition or -1 if the
 * topic-partition is not assigned. The current epoch is the epoch of the current owner.
 */
private BiFunction<Uuid, Integer, Integer> currentPartitionEpoch;

A member can only be given a partition once the previous owner's epoch shows it has released it. UNRELEASED_PARTITIONS is precisely "I've moved to the new epoch, but partition X is still owned by someone at an older epoch." This is the mechanism that makes the whole thing safe without a global barrier: instead of "everyone stops, then everyone starts," it's a per-partition handoff with an epoch-based lock.

The UNKNOWN case is a forward-compatibility fence. A coordinator downgrade encountering a member state written by a newer version can't reason about it, so it forcibly resets the member. Failing safe rather than guessing. Note it only throws if the member claims partitions — a member with nothing to lose is just re-reconciled from scratch.

9.4 Assignors

group-coordinator/.../assignor/:

The homogeneous/heterogeneous split is a real optimization: homogeneous lets you treat the assignment as balancing P partitions over M members with a stickiness bias, which is near-linear. Heterogeneous is a bipartite matching problem.

TopicIds and RangeSet exist to avoid materializing partition lists. A topic with 10,000 partitions is a RangeSet(0, 10000), not a HashSet of 10,000 Integers. UnionSet avoids copying when merging subscriptions. On a coordinator handling large groups these allocation-avoidance types dominate the profile.

TargetAssignmentBuilder computes the new target and diffs it against the old, emitting records only for members whose assignment changed — so a group of 1,000 where one member joins writes ~2 records, not 1,000.

Server-side assignment also means group.consumer.assignors is a broker config with a broker plugin interface. You can now ship a custom assignor without redeploying every consumer. That's the practical win people underrate.

9.5 Transaction coordinator and KIP-890

TransactionState (transaction-coordinator/.../TransactionState.java):

EMPTY           → Ongoing (AddPartitionsToTxn / AddOffsetsToTxn)
                → PrepareAbort (EndTxn abort, TV2 only)
ONGOING         → PrepareCommit (EndTxn commit) | PrepareAbort (EndTxn abort)
PREPARE_COMMIT  → CompleteCommit (all markers acked)
PREPARE_ABORT   → CompleteAbort  (all markers acked)
COMPLETE_COMMIT / COMPLETE_ABORT  → (evicted from cache)
DEAD            → transactionalId expired
PREPARE_EPOCH_FENCE → mid epoch bump, fencing older producers

State lives in __transaction_state (compacted, 50 partitions). The TransactionStateManager + TransactionCoordinator (still Scala in core/src/main/scala/kafka/coordinator/transaction/) own it. TxnMarkerQueue / TransactionMarkerChannelManager drive WriteTxnMarkers to every partition that participated.

Commit is two-phase:

1. Write PREPARE_COMMIT to __transaction_state, durably. (Now the outcome is decided.)
2. Send WriteTxnMarkers to every participating partition leader. Each appends a control batch (attributes bit 5) at the end of the partition, which advances that partition's LSO past the transaction's records.
3. When all markers are acked, write COMPLETE_COMMIT.

A coordinator crash between 1 and 3 is recovered by replaying __transaction_state and re-sending markers. Idempotent because a duplicate marker for an already-completed txn is a no-op.

KIP-890 (transaction version 2) fixes a genuine hanging-transaction bug class. In TV1, the producer's epoch was bumped only on InitProducerId, so a zombie producer that had been partitioned could still have an in-flight Produce land after its transaction was aborted — writing a record into a transaction that no longer existed, hanging the LSO forever.

TV2 fixes it by:

• Bumping the producer epoch on every transaction completion. A zombie's epoch is stale immediately, so its produce is fenced.
• Removing the explicit AddPartitionsToTxn round trip from the client — the broker implicitly adds the partition on first write, and verifies with the coordinator via a broker-side AddPartitionsToTxn (the VerificationGuard you see threaded through UnifiedLog.append).

VerificationGuard is the object identity token proving that a given append was verified against the coordinator for the current transaction. VerificationGuard.SENTINEL means "not applicable."

And KIP-1228 adds epoch validation on markers themselves — hence transactionVersion reaching all the way down into UnifiedLog.append:

// @param transactionVersion the transaction version for the records (1 for TV1, 2 for TV2, etc.)
//                           Used for epoch validation of transaction markers (KIP-1228).

The fact that a wire-protocol feature version has to be plumbed into the storage layer's append method tells you something about how deeply transactions cut across this system.

9.6 Share groups (KIP-932)

The headline feature of the 4.x line: queue semantics on top of the log. Many consumers can read the same partition cooperatively, with per-record acknowledgement — the RabbitMQ/SQS model, without giving up the log.

core/src/main/java/kafka/server/share/SharePartition.java. Per-record state:

Map.of((byte) 0,                     RecordState.ARCHIVED,      // represents gap
       AcknowledgeType.ACCEPT.id,    RecordState.ACKNOWLEDGED,
       AcknowledgeType.RELEASE.id,   RecordState.AVAILABLE,
       AcknowledgeType.REJECT.id,    RecordState.ARCHIVED)

States: AVAILABLE → ACQUIRED (delivered, lock timer running) → ACKNOWLEDGED (accepted) | ARCHIVED (rejected or delivery-count exhausted) | back to AVAILABLE (released or lock timeout).

private final NavigableMap<Long, InFlightBatch> cachedState;  // ConcurrentSkipListMap
private long startOffset;   // SPSO — share-partition start offset
private long endOffset;

startOffset (the Share-Partition Start Offset) is the analogue of a committed consumer offset, but it can only advance past a contiguous run of terminal-state records. Everything between startOffset and endOffset is in the in-flight map.

Mechanics you need to know:

• ShareFetch acquires records and starts an acquisition lock (group.share.record.lock.duration.ms). Lock expiry returns records to AVAILABLE — this is the redelivery mechanism.
• deliveryCount per record. Exceeding group.share.delivery.count.limit archives the record. With KIP-1191 (IBP_4_4_IV1) it goes to a dead-letter queue topic instead of being silently dropped.
• State is durable. The ShareCoordinator persists ShareSnapshot/ShareUpdate records into __share_group_state. PersisterStateBatchCombiner merges overlapping offset ranges to keep the state compact — this is the piece that determines whether the design scales, since naively you'd store per-offset state.
• GapWindow / persisterReadResultGapWindow handles the reconstruction problem: after restart, records that were compacted or never had state must be treated as gaps, not as unacknowledged.

What you give up: ordering. A share group has no ordering guarantee within a partition, because records are acknowledged out of order by design. If you need ordering, use a consumer group. Also, this is materially more per-record broker-side state than a consumer group's single offset — deliveryCompleteCount tracking exists specifically to make lag computable without walking the map.

Why this matters strategically: it removes the main reason teams run both Kafka and a traditional queue. The work-queue use case (competing consumers, per-message ack, redelivery, DLQ) was the one thing Kafka structurally couldn't do. Now it can, at the cost of a coordinator that has to track per-record state.

10. Log compaction

storage/.../log/{LogCleaner, LogCleanerManager, Cleaner, SkimpyOffsetMap}.java.

Selection. LogCleanerManager.grabFilthiestCompactedLog:

List<LogToClean> cleanableLogs = dirtyLogs.stream()
    .filter(ltc -> (ltc.needCompactionNow() && ltc.cleanableBytes() > 0)
                   || ltc.cleanableRatio() > ltc.log().config().minCleanableRatio)
    ...
LogToClean filthiest = cleanableLogs.stream().max(Comparator.comparingDouble(LogToClean::cleanableRatio))

cleanableRatio = dirtyBytes / (dirtyBytes + cleanBytes), compared against min.cleanable.dirty.ratio (default 0.5). needCompactionNow handles max.compaction.lag.ms — a hard deadline that forces compaction regardless of ratio, which exists for GDPR-style delete guarantees:

long maxCompactionLagMs = Math.max(log.config().maxCompactionLagMs, 0L);
long cleanUntilTime = now - maxCompactionLagMs;

There's also min.compaction.lag.ms at the other end — a floor, so recently-written records survive long enough for consumers to see them at least once.

The two-pass algorithm.

Pass 1 — build the offset map. Cleaner.buildOffsetMap scans the dirty section and inserts key → latestOffset into a SkimpyOffsetMap.

Pass 2 — copy segments. For each record in the clean+dirty range, retain it iff offsetMap.get(key) <= record.offset (i.e., this is the latest version) and it isn't an expired tombstone. Write survivors to a .cleaned segment, group several old segments into one new one (since they've shrunk), then .swap, then rename.

SkimpyOffsetMap is the interesting part. It is deliberately, named-in-the-class lossy-ish:

public SkimpyOffsetMap(int memory, String hashAlgorithm) {
    this.digest = MessageDigest.getInstance(hashAlgorithm);   // default MD5
    this.hashSize = digest.getDigestLength();                 // 16 bytes
    this.bytesPerEntry = hashSize + 8;                        // 24 bytes: hash + offset
    this.slots = memory / bytesPerEntry;
}

It stores the MD5 hash of the key, not the key. 24 bytes per entry regardless of key length. Linear probing on collision:

// limit attempt to number of slots once positionOf(..) enters linear search mode
int maxAttempts = slots + hashSize - 4;

The design trade: a 128-bit hash collision would cause the cleaner to retain the wrong record for a key. The probability at any realistic key count is negligible (birthday bound on 2^128), and the payoff is a fixed 24 bytes per key with no key materialization. MD5 here is a hash, not a security primitive — its cryptographic weaknesses are irrelevant and it's fast with a JDK intrinsic.

Capacity math you will need in production:

keys_per_cleaner_pass = (log.cleaner.dedupe.buffer.size / log.cleaner.threads)
                        / 24 bytes × log.cleaner.io.buffer.load.factor

Default: 128 MB / 1 thread / 24 × 0.9 ≈ 5 million keys per pass. If a partition's dirty section has more distinct keys than that, the cleaner compacts only a prefix of the dirty section per pass. It still makes progress, but slowly, and max.compaction.lag.ms guarantees will be missed. Symptom: max-dirty-percent stuck high, __consumer_offsets growing. Fix: raise log.cleaner.dedupe.buffer.size and/or log.cleaner.threads.

Tombstone retention. A null-valued record must be retained long enough for every consumer to observe the delete, otherwise a consumer that was offline would rebuild state with the key still present. delete.retention.ms (24h) governs. The mechanism is the delete horizon (attributes bit 6): when the cleaner first processes a batch containing tombstones, it stamps BaseTimestamp with the horizon time and sets the bit. Subsequent passes remove tombstones past that horizon. Prior to this bit, the horizon was inferred from segment modification time, which was fragile.

What compaction cannot do: it is not a delete-by-predicate, not transactional, and does not compact the active segment. CleanedTransactionMetadata handles the interaction with transactions — aborted-transaction records can be dropped, but their markers must be retained until every record they abort is gone.

log.cleanup.policy=compact,delete applies both: retention deletes old segments and compaction dedupes what remains. This is what __consumer_offsets uses.

11. Producer client internals

clients/src/main/java/org/apache/kafka/clients/producer/internals/.

send() [user thread]
  → interceptors → serializers → partitioner
  → RecordAccumulator.append()            ← batching happens here
  ...
[Sender thread — one per producer]
  loop:
    accumulator.ready(cluster, now)       → which nodes have sendable data
    accumulator.drain(...)                → node → List<ProducerBatch>
    build ProduceRequest per node
    client.poll()                         → NetworkClient / Selector

RecordAccumulator — ConcurrentMap<String /*topic*/, TopicInfo>, each with ConcurrentMap<Integer, Deque<ProducerBatch>>. Locking is per-deque:

Deque<ProducerBatch> dq = topicInfo.batches.computeIfAbsent(effectivePartition, k -> new ArrayDeque<>());
synchronized (dq) {
    if (partitionChanged(topic, topicInfo, partitionInfo, dq, nowMs, cluster)) continue;
    RecordAppendResult appendResult = tryAppend(timestamp, key, value, headers, callbacks, dq, nowMs);
    if (appendResult.appended()) return updatePartitionInfoOnAppend(...);
}

Note the while (true) { ... continue; } retry loop: the partition is peeked before taking the deque lock (to avoid holding a lock across partition selection), so it must be re-validated after. Classic optimistic pattern.

BufferPool.allocate is called outside the deque lock, and can block up to max.block.ms. The comment is explicit about why time.milliseconds() is refreshed after:

// NOTE: getting time may be expensive, so calling it under a lock should be avoided.

BufferPool maintains a free list of exactly-batch.size buffers plus a non-pooled remainder. A record larger than batch.size gets its own oversized buffer that is not returned to the pool. Consistently oversized records therefore mean constant allocation — one reason batch.size should be ≥ your p99 record size.

BuiltInPartitioner — the sticky partitioner, and it is not what people think. It is not "round-robin per batch." From updatePartitionInfo:

int producedBytes = partitionInfo.producedBytes.addAndGet(appendedBytes);
...
if (producedBytes >= stickyBatchSize && enableSwitch || producedBytes >= stickyBatchSize * 2) {
    // switch partition
}

It sticks to a partition until stickyBatchSize bytes have been produced to it, then switches. The * 2 upper bound is a forced switch even when enableSwitch is false:

// between stickyBatchSize and stickyBatchSize * 2 bytes, to better align with batch boundary

enableSwitch is false while the deque has incomplete batches, so the switch aligns with a batch boundary rather than splitting one. And nextPartition is load-aware — it weights by partition queue depth so a slow broker gets less traffic, rather than uniform round-robin.

Delivery timeout. deliveryTimeoutMs is the total budget from send() to callback, covering accumulator time + all retries. nextBatchExpiryTimeMs caches the earliest expiry so the Sender's poll timeout can be set correctly instead of scanning every deque each loop.

Batch splitting. splitAndReenqueue handles RecordTooLargeException after compression: the producer estimated the compressed size, the broker rejected it, so split in half and retry. Sequences are reassigned:

// We treat the newly split batches as if they are not even tried.
// We should track the newly created batches since they already have assigned sequences.

Ordering under retry. insertInSequenceOrder maintains the invariant that batches with assigned sequences leave the deque in sequence order:

// Further, once batches are being retried, we are reduced to a single in flight request for that
// partition. So when the subsequent batches come back in sequence order, they will have to be
// placed further back in the queue.

This, plus max.in.flight.requests.per.connection <= 5, plus the broker's 5-batch producer state retention, is the complete ordering-under-retry story for the idempotent producer.

ChunkedRecordAccumulator / ChunkedProducerBatch are newer: instead of one contiguous batch.size buffer per batch, allocate chunks. Reduces waste when partitions are numerous and batches rarely fill.

TransactionManager — client-side transaction state machine, sequence number allocation, and the TxnPartitionMap of per-partition sequence/epoch state. It is also what enforces that you can't call send() outside beginTransaction() in transactional mode.

12. Consumer client internals

Two implementations behind ConsumerDelegate:

• ClassicKafkaConsumer — the original. User thread does everything; a HeartbeatThread runs alongside for the classic protocol. ConsumerCoordinator implements JoinGroup/SyncGroup.
• AsyncKafkaConsumer — the new one (KIP-945), default with the consumer protocol.

The async design:

[user thread]                     [ConsumerNetworkThread]
poll()                            loop:
  → ApplicationEventQueue  ────►     process application events
  ◄──── BackgroundEventQueue         requestManagers.poll(now) → NetworkClientDelegate
  ← FetchBuffer                      networkClientDelegate.poll()
                                     → completed fetches into FetchBuffer

RequestManager implementations, each owning one concern:

ConsumerNetworkThread.poll() calls each manager's poll(now), gets back PollResults (requests to send + a next-poll timeout), and hands them to NetworkClientDelegate.

Why this matters: in the classic consumer, poll() had to be called frequently or the member was evicted, because heartbeats rode on the user thread's poll loop (partly — the classic protocol did have a background heartbeat thread, but rebalance callbacks still ran on the user thread and blocked everything). In the async consumer, network activity is fully decoupled: a slow poll() loop affects max.poll.interval.ms only.

ConsumerMembershipManager is the client mirror of the coordinator's reconciliation state machine. It receives a target assignment, invokes ConsumerRebalanceListener callbacks via ConsumerRebalanceListenerInvoker (on the user thread, dispatched through the background event queue), and only reports the partition as revoked in the next heartbeat once onPartitionsRevoked has returned. That's what makes the coordinator's "revocation proven by omission" protocol correct.

FetchBuffer / FetchCollector — the network thread parks completed fetches in FetchBuffer; poll() on the user thread drains and deserializes via FetchCollector. Deserialization on the user thread is deliberate: it keeps user-supplied Deserializer code (which can be slow or can throw) off the network thread.

CompletedFetch is where READ_COMMITTED filtering happens — it holds the aborted-transaction priority queue and drops records from aborted producers as it iterates.

AbstractStickyAssignor is still there for classic-protocol groups: CooperativeStickyAssignor implements incremental cooperative rebalancing (KIP-429), the pre-848 mitigation. If you're on the new protocol you don't need it.

13. Quotas, throttling, and backpressure

Four independent quota types, three enforcement mechanisms.

Types: produce byte-rate, fetch byte-rate, request percentage (request_percentage — CPU time in network+IO threads), and controller mutation rate (create/delete topics, partition changes).

ClientQuotaManager (server/src/main/java/org/apache/kafka/server/quota/) uses the metrics library's Rate over N sample windows (quota.window.num = 11, quota.window.size.seconds = 1). When a Sensor.record() throws QuotaViolationException, throttle time is computed as: how long must we pause so that the observed rate falls back to the quota, given the current window's accumulated value.

Quota entities resolve hierarchically: (user, client-id) → (user) → (client-id) → <default>. Configured in __cluster_metadata via ClientQuotaRecord, applied through ClientQuotasDelta.

Enforcement:

1. throttleTimeMs in the response. The client is expected to pause. Well-behaved clients do.

2. Channel muting. The broker also mutes the connection for throttleTimeMs via ThrottledChannel and the purgatory. From SocketServer:

   Try unmuting the channel. If there was no quota violation and the channel has not been throttled, it will be unmuted immediately. If the channel has been throttled, it will be unmuted only if the throttling delay has already passed by now.

   This is the part that makes quotas actually enforceable against a misbehaving client — you don't need the client's cooperation.

3. unrecordQuotaSensor. A subtlety worth reading:

   For a throttled fetch, the broker should return an empty response and thus should not record the value. Ideally, we would like to compute the throttle time before actually recording the value, but the current Sensor code couples value recording and quota checking very tightly. As a workaround, we will unrecord the value for the fetch in case of throttling.

   A negative record to undo an accounting entry, because the API couples measurement and decision. This is the sort of thing you find in mature code, and the honest comment is worth more than a clean-looking abstraction would be.

Replication throttling is separate: leader.replication.throttled.rate / follower.replication.throttled.rate plus per-topic *.throttled.replicas lists (validated by ThrottledReplicaListValidator). Used during reassignments so rebalancing doesn't starve production traffic.

Backpressure layers, from outermost in:

1. TCP receive window  ← MemoryPool exhaustion stops selector reads
2. newConnections queue (20/processor) ← acceptor blocks, kernel backlog absorbs
3. requestQueue (queued.max.requests=500) ← processors block on sendRequest
4. Quotas ← channel mute + throttleTimeMs
5. Purgatory ← delayed ops don't hold IO threads

Layer 5 is the one people miss. A purgatory-based design means a request that's waiting costs a DelayedOperation object and a timing-wheel entry, not a thread. That's why 8 I/O threads can serve 100,000 in-flight acks=all produce requests.

14. Comparison with other streaming systems

14.1 Kafka vs. Amazon Kinesis Data Streams

The single biggest architectural difference: Kinesis has no consumer-side log abstraction — it has a shard iterator.

The deep differences, not the feature table:

1. Kafka's offset is a number; Kinesis's iterator is a capability. A Kafka consumer stores an int64 and can resume from it a week later. A Kinesis ShardIterator expires in 5 minutes, so the consumer must store a SequenceNumber and re-derive an iterator. That's an extra API call and a latency floor per restart, and it means Kinesis can move data around behind the abstraction — which is exactly what makes resharding possible without rewriting history. Kafka's offset stability is why partition count can't decrease.

2. The shard lineage tree is Kinesis's price for elastic scaling. When you reshard, the parent shard is sealed and children are created. Consumers must read the parent to SHARD_END before reading children, or ordering breaks. KCL handles this; hand-rolled consumers routinely get it wrong. Kafka's answer to elasticity is "over-partition up front," which pushes the cost to provisioning time rather than runtime. Neither is obviously better; they're different failure modes.

3. Fan-out economics are inverted. In Kafka, adding a consumer group is nearly free — same page cache, same sendfile, incremental network. In Kinesis, the shared 2 MB/s egress means the n-th consumer group degrades the others, so past ~2 groups you're buying Enhanced Fan-Out at $0.015/consumer-shard-hour plus $0.013/GB. At 100 shards and 5 consumers that's ~$550/month in EFO hours alone before data. This is the number that drives most Kinesis→Kafka migrations.

4. KPL aggregation is a protocol-level wart. Because a record is capped at 1 MB and billed per record, the KPL packs many user records into one Kinesis record using a protobuf envelope. This is a client library convention, not a service feature — so any consumer not using KCL/KPL sees protobuf blobs. Kafka made batching part of the wire format and the storage format, so every consumer benefits and no one has to know.

5. Exactly-once. Kinesis has no equivalent of the idempotent producer or transactions. Building EOS on Kinesis means an external dedup store keyed by a producer-supplied ID, checked on the consumer side. That's a real system you have to build, operate, and scale.

When Kinesis is genuinely the right call: low-to-moderate, predictable throughput; you're all-in on AWS; you value zero operational burden over cost and features; your consumers are Lambda (the integration really is excellent). When it isn't: high fan-out, high throughput, long retention, exactly-once, compaction, or any need for a protocol other than AWS's.

14.2 Kafka vs. Apache Pulsar

Pulsar's core architectural claim is separation of serving and storage: brokers are stateless, storage is BookKeeper.

Honest assessment: Pulsar's storage separation is the better architecture on paper and Kafka's single-system operation is the better architecture in practice. Pulsar's instant failover and zero-data-movement rebalancing are real advantages that Kafka cannot match without a rewrite. But you operate three distributed systems instead of one, and BookKeeper's failure modes (ledger recovery, auto-recovery, bookie decommissioning) are their own specialty.

Note the convergence: share groups are Kafka's answer to Shared subscriptions, tiered storage answered offloaders, and KRaft answered "why do I need ZooKeeper." The remaining structural gap is storage separation.

14.3 Kafka vs. Redpanda

Redpanda is a C++ reimplementation of the Kafka protocol on Seastar (thread-per-core, shared-nothing, DPDK-capable, io_uring).

• Same wire protocol — your clients don't change.
• Thread-per-core with no shared state, so no locks and no cross-core cache-line traffic. Kafka's per-partition synchronized (lock) becomes a core-local operation.
• Raft per partition rather than ISR. Every partition is its own Raft group. This gives real quorum commit semantics instead of ISR, and eliminates the "ISR shrank to 1, now acks=all means nothing" class of problem.
• No JVM — no GC pauses, and memory is explicitly managed rather than page-cache-dependent. Redpanda bypasses the page cache and manages its own.
• fsync by default, unlike Kafka. Their argument is that with Raft + fast NVMe you can afford it; Kafka's counter-argument is the IOPS bill.

Their published p99 advantages are real and largely attributable to the thread-per-core model and the absence of GC. The counter-arguments: a much smaller ecosystem, per-partition Raft has higher metadata overhead at very large partition counts, and you're betting on one vendor's implementation of a protocol whose spec is "whatever Apache Kafka does."

The interesting lesson for a maintainer: most of Redpanda's advantage comes from things Kafka could have (thread-per-core partitioning of work, per-partition consensus) but can't adopt now without breaking everything. Architecture is path-dependent.

14.4 Kafka vs. WarpStream / diskless (KIP-1150)

WarpStream (and Confluent Freight, and AutoMQ) implement the Kafka protocol with S3 as the only storage layer. No local disks, no inter-AZ replication traffic, stateless agents.

The economics: inter-AZ transfer for replication is often the largest line item in a cloud Kafka bill — three copies means 2× your ingest crosses AZ boundaries at ~$0.02/GB each way. Writing straight to S3 (which is already multi-AZ) eliminates it entirely.

The cost: latency. S3 PUT is ~50–200 ms, so produce latency goes from single-digit ms to hundreds of ms. These systems batch aggressively and target workloads where a 200–500 ms end-to-end latency is acceptable — which, honestly, is most analytics pipelines.

KIP-1150 ("diskless topics") brings this model into Apache Kafka itself as a per-topic option. This is the most consequential thing on Kafka's roadmap: it makes the storage model a topic-level choice rather than a cluster-level architecture decision. Latency-sensitive topics stay on local disks with ISR; high-volume log/telemetry topics go diskless and cost 5–10× less to run.

14.5 Quick positioning of the rest

• Azure Event Hubs — has a Kafka-protocol endpoint; underneath it's its own system. Throughput Units, 7-day max retention (90 with Premium/Dedicated), no compaction, no transactions. Same managed-service trade as Kinesis with better protocol compatibility.
• Google Pub/Sub — not a log at all. Per-message ack, no offsets, no replay by position (Seek by timestamp only), automatic scaling, no partitions to manage. Pub/Sub Lite was the partitioned Kafka-like variant; it was deprecated.
• NATS JetStream — dramatically simpler operationally, single binary, excellent for edge/IoT/request-reply. Consumers have both push and pull modes and per-message ack. Not designed for the multi-TB-per-day replayable-log use case.
• Apache RocketMQ — architecturally closest to Kafka; dominant in China. Notable for a single shared CommitLog per broker with per-topic index files, rather than per-partition files — which makes it much better at very high topic counts and worse at per-partition sequential read locality.
• Redis Streams — a log data type inside Redis. Consumer groups with per-message ack (XACK), pending-entries list, XAUTOCLAIM for redelivery. Memory-bound, so retention is small. Excellent for a work queue at the scale where you already run Redis.

15. Design lessons: building your own

If you're implementing a durable log — or borrowing pieces for something else — these are the ideas in Kafka worth taking, ranked by leverage.

15.1 Make the batch, not the record, your atom

Compression, checksum, idempotence identity, transactional membership, index granularity, and replication unit should all be the same object. Kafka's per-record cost is ~6–8 bytes of framing and essentially zero CPU because everything expensive happens once per batch. Systems that made the record the atom (per-message compression, per-message ack, per-message index entry) spend 10–100× more per unit of data and can never claw it back.

Corollary: expose the batch to the client. If the producer builds the batch and the broker stores it byte-identically, you get zero-copy on both sides. If the broker has to reframe, you've lost.

15.2 Put the checksum boundary between client-authored and server-authored fields

PartitionLeaderEpoch outside the CRC is worth its awkwardness many times over. Before you finalize a format, list every field and mark who writes it. Any server-written field inside a client-computed checksum is a full-payload recompute on your hottest path.

15.3 Sparse index + linear scan beats a dense index

8 bytes per 4 KB of log. 2 MB of index per 1 GB segment. Compare to a B-tree or an LSM's block index

• bloom filters + manifest. The linear scan from the index hint is 4 KB of sequential page-cache reads — cheaper than the extra index levels you'd traverse to avoid it. This works because the access pattern is "seek once, then stream," which is true of almost every log consumer.

15.4 Design your data structures around the page cache, not around big-O

The warm-section binary search (§3.6) is the canonical example: same complexity, but a probe set that doesn't drift. Ask of every mmapped structure: which pages does a typical operation touch, and does that set change as the structure grows? If it changes, you have a periodic latency cliff you haven't found yet.

15.5 Don't fsync; replicate — but say so out loud

Kafka's durability model is "N page caches" not "one disk." That's a defensible engineering choice with a precisely-statable failure mode (simultaneous power loss to min.insync.replicas nodes). Whatever you choose, state the failure mode in the docs in one sentence, because your users will assume the other one.

15.6 Use a hierarchical timing wheel for timeouts, and put the priority queue over buckets

O(1) insert and delete matters when 99% of timers are cancelled. And the DelayQueue<TimerTaskList> trick — expensive structure over the cheap dimension (buckets), cheap structure over the expensive dimension (tasks) — generalizes far beyond timers.

15.7 Delayed operations should cost an object, not a thread

The purgatory is why 8 I/O threads serve 100,000 in-flight requests. Any time you're tempted to block a worker thread on a condition that might be satisfied by another request, build a watcher-list + timer instead. The complexity is real (see the deadlock comment) but the alternative is a thread pool sized by your worst-case concurrency.

15.8 One event queue, deterministic replay, deferred completion

The QuorumController pattern (§8.5). If your control plane state is small enough to fit in memory and your mutation rate is low enough for one core:

mutation → single-threaded handler → (records, response)
        → append records to replicated log
        → hold response until the append commits
        → apply records to in-memory state on commit
        → persistent (timeline) data structures so uncommitted state can be reverted

You get: no locks, deterministic replay, testable state machines, no dirty reads, and standby replicas for free. The cost is one core and the need to keep per-event work small. I'd take this trade in almost every control plane I've seen.

15.9 Level-triggered protocols beat edge-triggered ones

KIP-848's revocation is confirmed by a partition's absence from the next heartbeat, not by a revoke-ack message. A lost message costs a round trip, not correctness. Same idea in KRaft: brokers derive their own leader/follower duties from the shared log rather than receiving instructions the controller must track delivery of.

Rule of thumb: if your protocol has a message whose loss requires retry-tracking on the sender, consider whether the receiver could instead just report its state and let the sender diff.

15.10 Distinguish "fell behind" from "went away"

ELR (§7.5). Conflating them cost Kafka a decade of unclean-election data loss. Any time you evict a member from a quorum/ISR/membership set, ask whether the eviction reason tells you something about what that member knows.

15.11 Push the version boundary into the log

MetadataVersion + FeatureLevelRecord: the record that enables a feature is itself in the replicated log, so every node's behaviour is a pure function of its log prefix. No separate agreement protocol, no "wait for all nodes to restart," and downgrade safety is a property you can compute (didMetadataChange).

15.12 Backpressure at every layer, and let TCP be the outermost one

Kafka's five layers (§13) all terminate in "stop reading from the socket," which closes the receive window, which blocks the producer's send(). That's the only backpressure mechanism that can't be ignored by a misbehaving client. Everything else — response-carried throttle times, queue bounds — is an optimization on top of it.

15.13 Write the post-mortem in the comment

AbstractIndex's page-fault analysis, DelayedOperationPurgatory's 7-step deadlock, FileRecords's "avoid calling lastOffset()," ClientQuotaManager's "unrecord as a workaround because the Sensor API couples recording and checking." These comments are why a new maintainer can be productive in this codebase. The value isn't the explanation of what the code does — it's the record of what was tried and why it failed.

16. Maintainer's appendix

16.1 Build and test

./gradlew clean jar -PscalaVersion=2.13          # build
./gradlew processMessages                        # regenerate protocol classes from JSON
./gradlew unitTest                               # unit only
./gradlew :core:test --tests kafka.log.LogManagerTest
./gradlew :storage:test --tests '*UnifiedLog*'
./gradlew checkstyleMain spotbugsMain            # style/static analysis — CI will reject you otherwise
./gradlew :jmh-benchmarks:jmh -PjmhArgs='RecordBatchIterationBenchmark'
./gradlew :core:integrationTest                  # slow

test-common/ holds ClusterTestExtensions — the JUnit 5 machinery for spinning up real KRaft clusters in-process. @ClusterTest(types = {Type.KRAFT}, brokers = 3) gives you a real cluster in a unit test. This is the single most useful thing in the repo for writing a credible patch.

api-checker/ enforces public API compatibility. If you change a signature in clients, it will tell you.

16.2 Adding an RPC field — the actual workflow

1. Edit clients/src/main/resources/common/message/XxxRequest.json / XxxResponse.json. Bump validVersions. Add the field with "versions": "N+".
2. ./gradlew processMessages.
3. Add an ApiKeys entry if it's a new API.
4. Gate on MetadataVersion (or a dedicated feature) if brokers must agree.
5. Handle it in KafkaApis (or ControllerApis).
6. Add to ApiVersionsResponse handling if version negotiation matters.
7. Tests: MessageTest round-trip, RequestResponseTest, and an integration test.

The JSON is the source of truth. Never hand-edit generated classes.

16.3 Metrics you should be able to read cold

16.4 Debugging on a live broker

# Dump a segment, including producer/txn state
kafka-dump-log.sh --files 00000000000000000000.log --print-data-log --deep-iteration

# Just the batch headers — much faster, shows offsets/epochs/producerId/sequences
kafka-dump-log.sh --files 00000000000000000000.log

# Indexes, with consistency verification against the log
kafka-dump-log.sh --files 00000000000000000000.index --index-sanity-check
kafka-dump-log.sh --files 00000000000000000000.timeindex
kafka-dump-log.sh --files 00000000000000000000.txnindex

# Producer state snapshot
kafka-dump-log.sh --files 00000000000000000000.snapshot

# The metadata log itself
kafka-dump-log.sh --cluster-metadata-decoder --files __cluster_metadata-0/*.log

# Interactive metadata browser
kafka-metadata-shell.sh --snapshot __cluster_metadata-0/*.checkpoint
  > ls /topics
  > cat /topics/orders/0/data

# Quorum health
kafka-metadata-quorum.sh --bootstrap-controller localhost:9093 describe --status
kafka-metadata-quorum.sh --bootstrap-controller localhost:9093 describe --replication

# Feature levels
kafka-features.sh --bootstrap-server localhost:9092 describe

# Hanging transaction hunting
kafka-transactions.sh --bootstrap-server localhost:9092 list
kafka-transactions.sh --bootstrap-server localhost:9092 find-hanging --broker-id 1
kafka-transactions.sh --bootstrap-server localhost:9092 abort --topic t --partition 0 --start-offset N

The one to internalize: find-hanging + abort is the fix for a pinned LSO blocking READ_COMMITTED consumers. It's the most common exactly-once incident.

16.5 Failure modes worth memorizing

16.6 Reading order, if you're new to the tree

1. DefaultRecordBatch.java + DefaultRecord.java — the format. Everything else assumes it.
2. UnifiedLog.append — the write path in one method.
3. AbstractIndex — including the whole comment.
4. TimingWheel + DelayedOperationPurgatory — the async model.
5. Partition.scala — ISR, HW, epochs.
6. QuorumState.java + KafkaRaftClient class doc — consensus.
7. QuorumController.java — the control plane pattern.
8. CurrentAssignmentBuilder.java — the cleanest state machine in the repo.
9. RecordAccumulator + Sender — the client side.
10. jmh-benchmarks/ — what the maintainers actually consider hot.

16.7 Things that are not true anymore (but are still in every blog post)

• ZooKeeper. Gone in 4.0. kafka.zk, KafkaController, LeaderAndIsrRequest, UpdateMetadataRequest, StopReplicaRequest — all removed.
• org.apache.kafka.common.record.DefaultRecordBatch — moved to ...record.internal.
• The log layer is Scala. It's Java, in storage/.
• Message formats v0/v1 — removed. Only v2 (magic 2) is supported, so down-conversion is gone.
• --zookeeper on any CLI tool.
• The consumer rebalance protocol is JoinGroup/SyncGroup — that's "classic"; the default is KIP-848.
• Kafka can't do queues. Share groups (KIP-932) exist.
• unclean.leader.election.enable is your only availability lever — ELR (KIP-966) is the third option.
• Kafka requires local disks for everything — tiered storage (KIP-405) is GA, and diskless topics (KIP-1150) are in flight.

Compiled from apache/kafka @ 930ebc5608bb0ac938085321d09b402b850ca87b, 4.4.0-SNAPSHOT. Where I've quoted a comment, it's because the comment is the specification.