Distributed Log Architecture: Kafka Internals, Zero-Copy I/O, and Replication

1. The Core Engineering Challenge: Decoupled High-Throughput Streaming

Project Identity, Version & License Baseline

Traditional message brokers (e.g., RabbitMQ, ActiveMQ, IBM MQ) were designed around the JMS Queue abstraction:

1. Producers push messages into a queue.
2. The broker maintains in-memory tracking of individual message acknowledgments (ACK / NACK).
3. As soon as a consumer acknowledges a message, the broker deletes it from disk/memory.

+-------------------------------------------------------------------------------+
|                    Traditional Message Queue Bottleneck                       |
|                                                                               |
|  Producers ---> [ In-Memory Priority Queue ] ---> Consumers                   |
|                        |                                                      |
|             Individual Message Acks & Deletions                              |
|                                                                               |
|  Bottlenecks:                                                                 |
|  - Random disk deletions and index mutations degrade throughput under load.   |
|  - Slow consumers cause queue backlogs; as memory fills, brokers page to     |
|    disk, collapsing throughput precisely when load is highest.               |
|  - A message can only be consumed once; fan-out to 50 independent consumer   |
|    groups requires 50 separate copies of the data in memory.                  |
+-------------------------------------------------------------------------------+

The Distributed Commit Log, pioneered by Jay Kreps, Neha Narkhede, and Jun Rao at LinkedIn through Apache Kafka, inverted this model completely:

• Messages are never deleted upon consumption. The log is an append-only, immutable sequence of bytes.
• The broker maintains zero per-message acknowledgment state. Rather than tracking delivery receipts per record, consumers advance a single 64-bit integer—their current read Offset—persisted compactly per partition in the internal __consumer_offsets topic.
• As a result, 1 consumer or 1,000 independent consumers reading the same stream produce identical broker overhead.

2. Mental Model & Anatomy: The Broker & Partition Hierarchy

A production Kafka deployment organizes event streams through a strict physical hierarchy:

SYSTEM ARCHITECTURE BLUEPRINTDISTRIBUTED SYSTEMS

⤢Expand

ARCHITECTURAL FLOW · STEP 1 OF 5

1. Producer Batch Append to Leader Segment

Producer transmits a compressed RecordBatch. Leader broker appends records sequentially to active .log file. LEO increments to 150.

Architectural Invariant: Sequential Append: Bypasses random disk seeks by maintaining strictly immutable append-only commit segments.

ARCHITECTURAL FLOW · STEP 2 OF 5

2. Sparse Offset Indexing (.index)

Every 4KB of log data, Kafka inserts an 8-byte tuple (relative offset, physical byte offset) into memory-mapped .index file.

Architectural Invariant: mmap Efficiency: Keeps sparse indices memory-mapped in OS page cache; binary search finds physical byte position in microseconds.

ARCHITECTURAL FLOW · STEP 3 OF 5

3. In-Sync Replica (ISR) Fetch Pipeline

Follower replicas continuously dispatch Fetch requests. Follower 2 catches up to LEO 150. Follower 3 lags temporarily at 145.

Architectural Invariant: ISR Invariant: Replicas remain in the ISR set only if they fetch within replica.lag.time.max.ms (default 30s).

ARCHITECTURAL FLOW · STEP 4 OF 5

4. High Watermark (HW) Barrier Advance

Once Follower 3 catches up to 150, the High Watermark advances to 150. Producer receives acknowledgement (acks=all). Consumers can now read records up to HW.

Architectural Invariant: Safety Barrier: Consumers NEVER observe records past the High Watermark, guaranteeing no phantom dirty reads on leader crash.

ARCHITECTURAL FLOW · STEP 5 OF 5

5. Zero-Copy Consumer Read (Linux sendfile)

Consumer requests records. Kafka issues sendfile() syscall. Data streams directly from OS Page Cache to NIC DMA buffer with zero JVM memory copies.

Architectural Invariant: Zero-Copy Guarantee: Eliminates user-space buffer copies and context switches, saturating 40GbE network interfaces with near-zero CPU usage.

1. Topic: A logical stream category (e.g., orders.v1, telemetry.metrics).
2. Partition: The fundamental unit of parallelism and ordering. A topic is split into $N$ partitions distributed across a cluster of brokers. Messages with the same key (e.g., user_id = "1042") route to the same partition, guaranteeing strict total ordering per key.
3. Log Segment: On the broker’s physical filesystem, a partition is divided into 1 GB contiguous Segments.
4. In-Sync Replicas (ISR): For high availability, each partition is replicated across $M$ brokers. One broker acts as the Partition Leader (handling all reads and writes), while the remaining brokers act as Followers continuously pulling data.

3. On-Disk Segment Storage Engine

A common interview myth is that “Kafka is fast because it keeps all messages in memory.”

Kafka writes every single message to persistent storage immediately (appending into the OS page cache backed by sequential segment files). It achieves multi-gigabyte throughput because it exploits the physical storage realities explored in CORE-M04-001: sequential disk I/O ($600\text{ MB/s}-7\text{ GB/s}$ on RAID arrays and NVMe SSDs) outperforms random disk seeks by $1,000\times$ to $10,000\times$, and avoids the CPU cache misses and GC overhead of random pointer-chasing in a managed JVM heap.

Inside a Partition Directory

On the broker’s disk, a partition directory (/var/lib/kafka/data/orders-0/) contains a series of segment files named after the segment’s starting offset:

/var/lib/kafka/data/orders-0/
├── 00000000000000000000.log         <- Actual message payloads
├── 00000000000000000000.index       <- Sparse offset index
├── 00000000000000000000.timeindex   <- Timestamp-to-offset index
├── 00000000000010485760.log         <- Next active segment (Starts at offset 10,485,760)
├── 00000000000010485760.index
└── leader-epoch-checkpoint

1. The Data File (.log)

The .log file stores contiguous RecordBatches. Each batch contains:

• baseOffset (64-bit integer): First logical offset in the batch.
• batchLength (32-bit integer).
• partitionLeaderEpoch (32-bit integer): Prevents log truncation anomalies during failover.
• magic (Byte): Record format version.
• crc32 (32-bit checksum): Detects bit rot on physical media.
• records: Array of individual records with relative offset, timestamp, key, value, and headers.

Producers append to the end of the file sequentially ($O(1)$ disk append).

2. The Sparse Offset Index (.index)

If Kafka maintained an index entry for every single message, the index file would consume gigabytes of memory. Instead, Kafka uses a Sparse Index:

• It records an entry only once every 4 KB of data (index.interval.bytes = 4096).
• Each entry is a fixed 8-byte tuple: [relative_offset (4 bytes), physical_position (4 bytes)].

Target: Read Offset 1045
---------------------------------------------------------------------------
1. Binary search the .index file (memory-mapped in RAM):
   [ Offset 0   -> Byte 0     ]
   [ Offset 500 -> Byte 4210  ]
   [ Offset 1000-> Byte 8450  ]  <-- Highest offset <= 1045
   [ Offset 1500-> Byte 12890 ]

2. Seek directly to physical Byte 8450 in the .log file.
3. Scan sequentially forward from Byte 8450 until Offset 1045 is reached!
---------------------------------------------------------------------------
Result: Exact message located with zero disk seek thrashing.

The entire .index file is memory-mapped (mmap) directly into the operating system’s page cache, allowing the binary search to execute entirely in RAM.

4. Zero-Copy Network I/O (sendfile)

When a consumer requests messages (FetchRequest(offset=1045, max_bytes=1MB)), how does the broker transmit data from disk to the network card?

The Traditional 4-Copy Bottleneck

In a traditional application server (e.g., standard Tomcat, Node.js, or Python web servers), transmitting a file across a socket requires four context switches and four memory copies:

[ Disk Media ]
     | (1. Hardware DMA transfer)
     v
[ OS Kernel Page Cache ]
     | (2. CPU memory copy: Kernel Space -> User Space)
     v
[ JVM Application Memory ] (User Space buffer)
     | (3. CPU memory copy: User Space -> Socket Buffer)
     v
[ OS Kernel Socket Buffer ]
     | (4. Hardware DMA transfer)
     v
[ Network Interface Card (NIC) ]

This traditional pipeline burns massive CPU cycles copying identical bytes back and forth across the user-kernel boundary and pollutes CPU caches.

The Kafka Zero-Copy Path (sendfile)

Kafka bypasses user-space memory entirely using the Linux kernel’s sendfile() system call (exposed via Java’s FileChannel.transferTo()):

[ Disk Media ]
     | (1. Hardware DMA transfer)
     v
[ OS Kernel Page Cache ] ===============================+
                                                        | (2. Direct Hardware DMA transfer)
                                                        v
                                          [ Network Interface Card (NIC) ]

1. The broker passes the open file descriptor and the socket descriptor directly to the operating system kernel.
2. The kernel instructs the DMA controller to stream bytes directly from the OS Page Cache to the NIC ring buffer.
3. Zero bytes enter the JVM user-space memory.
4. Zero CPU cycles are wasted on memory copies.
5. Zero JVM Garbage Collection pauses occur.

This architectural insight explains why Kafka brokers typically run with small JVM heaps (e.g., 6 GB–8 GB), leaving the remaining 60 GB–120 GB of server RAM entirely to the Linux OS Page Cache. Once hot segments reside in the Page Cache, traditional user-space socket writes cross the DRAM memory bus 3 times per byte served (Page Cache $\to$ JVM Buffer $\to$ Socket Buffer $\to$ NIC DMA), whereas sendfile() scatter-gather DMA reads directly from Page Cache $\to$ NIC (1 DRAM bus traversal, a $66.7\%$ reduction in memory-bus bandwidth).

5. Replication, Quorums, and the High Watermark

Kafka ensures high availability and zero data loss through In-Sync Replica (ISR) Quorums:

Topic: orders, Partition: 0, Replication Factor: 3
+---------------------------------------------------------------------------------+
| Leader Broker (Broker 1):    [0] [1] [2] [3] [4] [5] [6] [7]  (LEO = 8)         |
|                                                                                 |
| Follower A (Broker 2 - ISR): [0] [1] [2] [3] [4] [5] [6] [7]  (LEO = 8)         |
|                                                                                 |
| Follower B (Broker 3 - ISR): [0] [1] [2] [3] [4] [5]          (LEO = 6)         |
+---------------------------------------------------------------------------------+
                           High Watermark (HW) = 6

Key Terminology

• Log End Offset (LEO): The offset of the next record to be written in a partition on a specific broker. In the diagram above, the Leader’s LEO is 8 (records 0..7 are stored locally).
• In-Sync Replicas (ISR): The set of replicas (including the leader) that are actively caught up with the leader. If a follower fails to send a FetchRequest or catch up to the leader’s LEO within replica.lag.time.max.ms (default: 30 seconds), the leader shrinks the ISR set by evicting the lagging follower.
• High Watermark (HW): Formally defined as the minimum LEO across all active members of the ISR:

In the diagram above, Follower B has replicated offsets 0..5 (${\text{LEO}}_B=6$), so $\text{HW}=\min (8,8,6)=6$, exposing only committed offsets $0\le \text{offset}<\text{HW}$ (0..5) to consumers. Under acks=all and unclean.leader.election.enable=false, a write is accepted only when the active quorum satisfies:

The Consumer Isolation Invariant

Consumers are strictly forbidden from reading messages beyond the High Watermark.

Why? If a consumer were allowed to read offsets 6 and 7 from the leader, and the leader immediately suffered a hardware power failure, Follower B (whose LEO is 6) might be elected as the new leader. Offsets 6 and 7 would vanish from the cluster! The consumer would have processed phantom records that never permanently existed.

The Durability Matrix (acks & min.insync.replicas)

Durability is configured via the client producer’s acks parameter combined with the broker’s min.insync.replicas:

CORRECTNESS INVARIANT:
If acks=all and the number of alive ISR brokers drops below min.insync.replicas:
The leader broker REJECTS all incoming produce requests with NotEnoughReplicasException!
It chooses AVAILABILITY LOSS over DATA LOSS.

6. End-to-End Exactly-Once Semantics (EOS)

Achieving Exactly-Once Semantics (EOS) across a distributed pipeline requires solving two distinct problems:

1. Idempotent Producer (Single Partition)

When a producer retries due to a network timeout, the broker must not append a duplicate:

• Each producer is assigned a 64-bit Producer ID (PID) by the broker.
• For each partition, the producer maintains a monotonically increasing Sequence Number.
• The broker caches the last 5 sequence numbers per PID. If an incoming message carries a sequence number $\le$ the recorded sequence number, the broker acknowledges the write but silently discards the duplicate payload.

2. Transactional Coordinator (Atomic Cross-Partition Multi-Topic Writes)

In a stream processing application (read-process-write):

1. Consumer reads input record from Topic A (Offset 50).
2. Application computes state update.
3. Producer writes result to Topic B (Offset 200).
4. Application commits consumer offset 50 to __consumer_offsets.

If the service crashes between step 3 and step 4, the record will be re-processed on reboot, emitting duplicate events into Topic B.

The Kafka Transaction Protocol: Kafka introduces a Transaction Coordinator broker and a special topic __transaction_state:

• The producer initiates a transaction: beginTransaction().
• Both the business message (Topic B) and the consumer offset commit (__consumer_offsets) are written under a common transaction ID.
• The coordinator writes a two-phase commit marker: PREPARE_COMMIT $\to$ COMMIT.
• Downstream consumers configured with isolation.level = read_committed filter out uncommitted and aborted messages, observing only atomically committed records.

7. Consumer Group Coordination & Rebalance Protocols

Horizontal scalability on the consumer side is achieved via Consumer Groups:

• A topic with 12 partitions can be processed in parallel by up to 12 consumer instances in the same group.
• If there are 3 consumers, each processes 4 partitions.
• If a consumer crashes, the group must Rebalance to reassign the orphaned partitions.

Topic: telemetry (4 Partitions)

Initial State: 2 Consumers
[ Partition 0 ] ---> Consumer 1
[ Partition 1 ] ---> Consumer 1
[ Partition 2 ] ---> Consumer 2
[ Partition 3 ] ---> Consumer 2

Consumer 3 Joins the Group -> REBALANCE TRIGGERED!
[ Partition 0 ] ---> Consumer 1
[ Partition 1 ] ---> Consumer 2
[ Partition 2 ] ---> Consumer 3
[ Partition 3 ] ---> Consumer 3

Eager Rebalance vs. Incremental Cooperative Rebalance

Historically, Kafka used the Eager Rebalance Protocol:

1. When any consumer joined or left, the Group Coordinator ordered all consumers to revoke all partition assignments.
2. All consumption across the entire group stopped dead (Stop-the-World pause).
3. All consumers rejoined, and the leader reassigned partitions from scratch.
4. On large clusters (1,000+ partitions), eager rebalances took minutes, causing massive consumer lag spikes.

The Solution: Incremental Cooperative Rebalancing (KIP-429):

• Consumers retain partitions that are not moving.
• Only the specific partitions that must migrate are revoked and reassigned across two coordinated rounds.
• Unaffected partitions continue processing data continuously with zero downtime.

8. Mathematical Capacity Planning: Cluster Sizing

When sizing a production Kafka cluster, capacity planning is driven by three physical limits: Storage Volume, Network Bandwidth, and Partition Count.

1. Storage Volume Formula

Where $R_{\text{ingest}}$ is the cluster-wide uncompressed (or post-producer-compression) ingestion rate in $\text{MB/s}$, $T_{\text{retention}}$ is the retention window in seconds, $\text{RF}$ is the Replication Factor (typically $3$), and ${\alpha }_{\text{headroom}}=0.25$ represents a $25\%$ operational safety margin for .index/.timeindex files, log compaction slack, and partition rebalance buffers.

Worked Scenario (using SI decimal storage units, $1\text{ TB}=10^6\text{ MB}$):

• Cluster ingestion rate $R_{\text{ingest}}$: $100\text{ MB/s}$ ($800\text{ Mbps}$).
• Retention window $T_{\text{retention}}$: $7\text{ days}=7\times 86,400=604,800\text{ seconds}$.
• Replication Factor $\text{RF}$: $3$, Headroom factor: $1.25$.

Across a balanced $B=6$-broker cluster, each broker requires $\frac{226.8\text{ TB}}{6}=37.8\text{ TB}$ of local NVMe storage.

2. Network Interface Saturation Formula

Network bandwidth on Kafka brokers is strongly asymmetric because leaders must serve both inter-broker follower replication ($\text{RF}-1$ followers) and independent consumer group fan-out ($G_{\text{consumers}}$):

Where ${\sigma }_{\text{skew}}\ge 1.0$ captures partition leadership or hot-key traffic skew across $B$ brokers.

• For cluster ingress $R_{\text{ingest}}=100\text{ MB/s}$, $\text{RF}=3$, and $G_{\text{consumers}}=4$ independent consumer groups:

• Balanced cluster (${\sigma }_{\text{skew}}=1.0,B=6$): Each broker transmits $E_{\text{broker}}=100\text{ MB/s}$ ($0.8\text{ Gbps}$).
• Hot-key / skewed leader worst case: If a single broker hosts a hot partition receiving the full $100\text{ MB/s}$ ingress (or per-broker ingress itself is $100\text{ MB/s}$), that single broker’s NIC egress reaches $600\text{ MB/s}$ ($4.8\text{ Gbps}$), saturating a $1\text{ GbE}$ link $4.8\times$ over and requiring at least a $10\text{ GbE}$ (or $25\text{ GbE}$) NIC to prevent follower ISR evictions during consumer catch-up bursts.

9. Operational Failure Modes & Production Troubleshooting

Failure Mode 1: The “Poison Pill” Consumer Lag Explosion

Symptom: Consumer group lag climbs steadily on Partition 3. Other partitions process normally. CPU on the consumer instance hits 100%.
Root Cause: A malformed payload (“poison pill”) at offset 98,201 causes an unhandled deserialization or business logic exception. The consumer crashes, restarts, fetches the same offset 98,201, and crashes again in an infinite loop.
Resolution:

1. Implement a Dead Letter Queue (DLQ) pattern: catch fatal deserialization errors, publish the malformed byte payload and stack trace to an orders.dlq topic, and commit the offset to proceed.
2. Monitor consumer group lag via Prometheus kafka_consumergroup_lag.

Failure Mode 2: Under-Replicated Partitions (URP)

Symptom: The metric UnderReplicatedPartitions > 0 fires an alert.
Root Cause: One or more follower brokers have fallen behind the leader and dropped out of the ISR. Common triggers:

• Disk I/O saturation on a follower (e.g., long fsync stalls).
• Garbage collection pause on a broker JVM exceeding replica.lag.time.max.ms.
• Network link degradation between availability zones.
  Resolution:

1. Check kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions.
2. Inspect broker system logs for disk I/O latency spikes or network timeouts.
3. If a broker disk fails, the remaining ISR replicas continue serving traffic safely. Replace the failed node and execute kafka-reassign-partitions.sh.

10. Concrete Implementation: In-Memory Distributed Log Engine

The following complete TypeScript implementation models the core architectural mechanics of an append-only Kafka broker: sequential segment logging, sparse offset indexing with binary search, ISR replica tracking, and High Watermark reader barriers:

import { createHash } from 'node:crypto';

export interface RecordMessage {
  offset: number;
  timestamp: number;
  key: string;
  payload: string;
  checksum: string;
}

export interface SparseIndexEntry {
  relativeOffset: number;
  physicalIndex: number;
}

/**
 * An individual 1GB-equivalent on-disk log segment
 */
export class LogSegment {
  public readonly baseOffset: number;
  public records: RecordMessage[] = [];
  public index: SparseIndexEntry[] = [];
  private readonly indexInterval: number; // Sparse interval (e.g. Every 4 records)

  constructor(baseOffset: number, indexInterval = 4) {
    this.baseOffset = baseOffset;
    this.indexInterval = indexInterval;
  }

  append(key: string, payload: string, offset: number): RecordMessage {
    const checksum = createHash('sha256').update(`${offset}:${key}:${payload}`).digest('hex').slice(0, 8);
    const record: RecordMessage = {
      offset,
      timestamp: Date.now(),
      key,
      payload,
      checksum,
    };

    const physicalIndex = this.records.length;
    this.records.push(record);

    // Populate sparse index every N entries
    if (physicalIndex % this.indexInterval === 0) {
      this.index.push({
        relativeOffset: offset - this.baseOffset,
        physicalIndex,
      });
    }

    return record;
  }

  /**
   * Fast lookup using sparse index binary search followed by sequential scan
   */
  readFrom(targetOffset: number, maxRecords: number): RecordMessage[] {
    if (this.records.length === 0 || targetOffset < this.baseOffset) return [];

    const relativeTarget = targetOffset - this.baseOffset;

    // 1. Binary search the sparse index to find the nearest physical starting position
    let low = 0;
    let high = this.index.length - 1;
    let candidateIndex = 0;

    while (low <= high) {
      const mid = Math.floor((low + high) / 2);
      if (this.index[mid].relativeOffset <= relativeTarget) {
        candidateIndex = this.index[mid].physicalIndex;
        low = mid + 1;
      } else {
        high = mid - 1;
      }
    }

    // 2. Sequential forward scan from sparse index point to exact target offset
    const results: RecordMessage[] = [];
    for (let i = candidateIndex; i < this.records.length; i++) {
      if (this.records[i].offset >= targetOffset) {
        results.push(this.records[i]);
        if (results.length >= maxRecords) break;
      }
    }

    return results;
  }
}

/**
 * Replicated Partition Engine with ISR tracking and High Watermark barrier
 */
export class ReplicatedPartition {
  public readonly topic: string;
  public readonly partitionId: number;
  private segments: LogSegment[] = [];
  private activeSegment: LogSegment;
  private readonly segmentCapacity: number;
  private nextOffset = 0;

  // Replication State
  public leaderEpoch = 1;
  public isrBrokers = new Set<string>(['broker-1', 'broker-2', 'broker-3']);
  private followerLEOs = new Map<string, number>([
    ['broker-2', 0],
    ['broker-3', 0],
  ]);
  public highWatermark = 0;

  constructor(topic: string, partitionId: number, segmentCapacity = 8) {
    this.topic = topic;
    this.partitionId = partitionId;
    this.segmentCapacity = segmentCapacity;
    this.activeSegment = new LogSegment(0);
    this.segments.push(this.activeSegment);
  }

  /**
   * Producer Write Path: Appends to leader log
   */
  produce(key: string, payload: string, minInsyncReplicas = 2): { offset: number; hw: number } {
    if (this.isrBrokers.size < minInsyncReplicas) {
      throw new Error(`NotEnoughReplicasException: ISR size ${this.isrBrokers.size} < min.insync.replicas ${minInsyncReplicas}`);
    }

    const assignedOffset = this.nextOffset++;

    // Roll new segment if active segment exceeds capacity
    if (this.activeSegment.records.length >= this.segmentCapacity) {
      this.activeSegment = new LogSegment(assignedOffset);
      this.segments.push(this.activeSegment);
    }

    this.activeSegment.append(key, payload, assignedOffset);
    return { offset: assignedOffset, hw: this.highWatermark };
  }

  /**
   * Follower Fetcher: Followers replicate from leader and advance HW
   */
  followerFetch(brokerId: string, followerFetchOffset: number): { records: RecordMessage[]; hw: number } {
    if (!this.isrBrokers.has(brokerId)) {
      throw new Error(`Broker '${brokerId}' is not in the active ISR set`);
    }

    // Update follower LEO
    this.followerLEOs.set(brokerId, followerFetchOffset);

    // Recompute High Watermark: min(LEO of all active ISR members)
    const activeFollowerLEOs = Array.from(this.followerLEOs.entries())
      .filter(([id]) => this.isrBrokers.has(id))
      .map(([, leo]) => leo);
    const allLEOs = [this.nextOffset, ...activeFollowerLEOs];
    this.highWatermark = Math.min(...allLEOs);

    // Return records up to the requested offset
    const records = this.readInternal(followerFetchOffset, 4);
    return { records, hw: this.highWatermark };
  }

  /**
   * Consumer Read Path: STRICTLY bounded by the High Watermark (Zero dirty reads)
   */
  consumerFetch(requestedOffset: number, maxRecords = 10): RecordMessage[] {
    if (requestedOffset >= this.highWatermark) {
      return []; // Return empty if consumer tries to read beyond High Watermark
    }

    const records = this.readInternal(requestedOffset, maxRecords);
    // Filter out anything beyond high watermark
    return records.filter((r) => r.offset < this.highWatermark);
  }

  private readInternal(offset: number, maxRecords: number): RecordMessage[] {
    // Locate the starting segment containing the offset
    let startSegIdx = -1;
    for (let i = this.segments.length - 1; i >= 0; i--) {
      if (this.segments[i].baseOffset <= offset) {
        startSegIdx = i;
        break;
      }
    }
    if (startSegIdx === -1) return [];

    // Read across segment boundaries if maxRecords spans multiple segments
    const collected: RecordMessage[] = [];
    let nextTarget = offset;
    for (let i = startSegIdx; i < this.segments.length && collected.length < maxRecords; i++) {
      const chunk = this.segments[i].readFrom(nextTarget, maxRecords - collected.length);
      collected.push(...chunk);
      if (chunk.length > 0) {
        nextTarget = chunk[chunk.length - 1].offset + 1;
      }
    }
    return collected;
  }
}

function runDemo(): void {
  console.log("=== Kafka Commit Log & Partition Replication Architecture ===");
  const partition = new ReplicatedPartition("orders-events", 0);
  console.log(`Initialized topic '${partition.topic}', partition ${partition.partitionId} with ISR [${Array.from(partition.isrBrokers).join(", ")}]`);
  const produce1 = partition.produce("key-order-1", "OrderPlaced: { id: 1001, amount: 250 }");
  const produce2 = partition.produce("key-order-2", "OrderPaid: { id: 1001, status: 'PAID' }");
  console.log(`Produced record 1 at offset ${produce1.offset}, partition highWatermark: ${produce1.hw}`);
  console.log(`Produced record 2 at offset ${produce2.offset}, partition highWatermark: ${produce2.hw}`);

  // Round 1: Followers pull uncommitted records starting at offset 0
  const b2Pull = partition.followerFetch("broker-2", 0);
  const b3Pull = partition.followerFetch("broker-3", 0);
  console.log(`Followers pulled ${b2Pull.records.length} & ${b3Pull.records.length} records; subsequent fetch confirms LEO=2...`);

  // Round 2: Followers request next offset 2, confirming local persistence up to LEO=2
  partition.followerFetch("broker-2", 2);
  partition.followerFetch("broker-3", 2);
  console.log(`Followers caught up! High Watermark advanced to: ${partition.highWatermark}`);

  const consumed = partition.consumerFetch(0, 10);
  console.log(`Consumer fetched ${consumed.length} committed records up to HW:`);
  for (const rec of consumed) {
    console.log(`  Offset ${rec.offset}: key='${rec.key}', payload='${rec.payload}'`);
  }
}

runDemo();

Simulation Execution Output & State Transitions

=== Kafka Commit Log & Partition Replication Architecture ===
Initialized topic 'orders-events', partition 0 with ISR [broker-1, broker-2, broker-3]
Produced record 1 at offset 0, partition highWatermark: 0
Produced record 2 at offset 1, partition highWatermark: 0
Followers pulled 2 & 2 records; subsequent fetch confirms LEO=2...
Followers caught up! High Watermark advanced to: 2
Consumer fetched 2 committed records up to HW:
  Offset 0: key='key-order-1', payload='OrderPlaced: { id: 1001, amount: 250 }'
  Offset 1: key='key-order-2', payload='OrderPaid: { id: 1001, status: 'PAID' }'

11. Design Alternatives & Trade-offs: Kafka vs. RabbitMQ vs. Pulsar vs. Redpanda

12. Summary & Architectural Decision Guide

+-----------------------------------------------------------------------------------+
|                        Distributed Log Architecture Matrix                        |
+--------------------------+-----------------------+--------------------------------+
| Architecture Feature     | Design Choice         | Production Rationale           |
+--------------------------+-----------------------+--------------------------------+
| Storage Layout           | Append-Only Segments  | Converts random mutations to   |
|                          | with Sparse Index     | sequential streaming I/O       |
+--------------------------+-----------------------+--------------------------------+
| Network Transport        | Linux sendfile()      | Zero-copy DMA bypasses JVM     |
|                          | (Page Cache to NIC)   | user-space and eliminates GC   |
+--------------------------+-----------------------+--------------------------------+
| Replication Quorum       | In-Sync Replicas (ISR)| Dynamic quorum balancing       |
|                          | with High Watermark   | zero data loss and latency     |
+--------------------------+-----------------------+--------------------------------+
| Scaling Abstraction      | Partitioned Ordering  | Strict FIFO ordering per key   |
|                          | by Hash(Key)          | with horizontal scale-out      |
+--------------------------+-----------------------+--------------------------------+

The Streaming Architect’s Rule of Thumb

1. Never Size Partitions Arbitrarily: Keep partition counts well within KRaft metadata and file-descriptor budgets (typically $<4,000$ partitions per broker) to prevent .index mmap fragmentation and slow failover recovery.
2. Always Pair acks=all with min.insync.replicas=2 (and RF=3): Setting acks=all with min.insync.replicas=1 provides zero redundancy if a solitary in-sync leader fails immediately after local flush.
3. Let the Operating System Manage Memory: Keep Kafka broker JVM heaps compact (4 GB–8 GB) and dedicate $80\%+$ of physical RAM to the OS Page Cache for zero-copy sendfile() throughput.

13. Guided Design Exercises

Exercise 1: Quantitative Broker Storage, Segment File Descriptors & NIC Egress Sizing

A payments platform ingests $R_{\text{ingest}}=150\text{ MB/s}$ (post-LZ4 compression) into a 9-broker Kafka cluster with replication factor $\text{RF}=3$, retention window $T_{\text{retention}}=14\text{ days}$, operational storage headroom ${\alpha }_{\text{headroom}}=0.25$, segment size segment.bytes = 1073741824 ($1\text{ GB}$), and $G_{\text{consumers}}=5$ independent consumer groups.

1. Calculate the total cluster storage footprint (in decimal TB) and the per-broker NVMe storage requirement.
2. Calculate how many on-disk segment triples (.log, .index, .timeindex) and open file descriptors exist per broker.
3. Compute the per-broker NIC egress bandwidth (in MB/s and Gbps) under balanced leadership (${\sigma }_{\text{skew}}=1.0$) and under a $1.8\times$ hot-partition skew (${\sigma }_{\text{skew}}=1.8$), and determine whether $10\text{ GbE}$ NICs are sufficient.

Exercise 2: High Watermark, Leader Epoch Truncation & Durability Quorum Design

A financial ledger topic is configured with $\text{RF}=3$ (broker-1 as leader, broker-2 and broker-3 as followers), min.insync.replicas = 2, and unclean.leader.election.enable = false.

1. At $t_0$, broker-1 has ${\text{LEO}}_1=500$, broker-2 has ${\text{LEO}}_2=500$, and broker-3 lags at ${\text{LEO}}_3=480$ (still inside replica.lag.time.max.ms). What is the partition High Watermark ($\text{HW}$), and which offsets are visible to consumers?
2. At $t_1$, broker-3 stalls in a 40-second disk fsync freeze and is evicted from the ISR (replica.lag.time.max.ms = 30000). Immediately upon eviction, how does $\text{HW}$ change, and why is this transition safe?
3. At $t_2$, while broker-3 is still out of the ISR, broker-2 suffers a power outage. Trace what happens to incoming producers with acks=all versus existing consumers reading offsets $<500$.

Exercise 3: Catch-Up Consumer Page Cache Pollution & Follower ISR Starvation Drill

During a routine deployment, a historical analytics consumer group restarts and begins backfilling 48 hours of log segments ($25\text{ TB}$ cold on NVMe) at maximum throughput across all 9 brokers. Within 90 seconds, UnderReplicatedPartitions spikes from $0$ to $140$, and live producers experience $p99$ latency surges from $8\text{ ms}$ to $950\text{ ms}$.

1. Explain the kernel-level mechanics of why cold consumer reads evict active tail segments from the Linux OS Page Cache and starve follower ReplicaFetcherThread I/O.
2. Specify the exact broker-level quota commands, fadvise / tiered storage settings, and fetch throttling configurations required to isolate live ISR replication from cold catch-up consumers.

14. Authoritative References & Version Context

• Version Baseline: Architectural details, Cooperative Sticky rebalancing, and KRaft metadata quorums align with Apache Kafka v3.7+ and v4.0 (Apache License 2.0), where KRaft (KIP-500 / KIP-833) completely replaced Apache ZooKeeper.
• Kreps, J., Narkhede, N., & Rao, J. (2011). Kafka: a Distributed Messaging System for Log Processing. Proceedings of NetDB 2011.
• Kreps, J. (2013). The Log: What every software engineer should know about real-time data’s unifying abstraction. LinkedIn Engineering Blog.
• Apache Kafka Project Documentation. Apache Kafka Design & Replication Internals.
• Apache Kafka KIP-500. Replace ZooKeeper with a Self-Managed Metadata Quorum (KRaft).
• Linux Programmer’s Manual. sendfile(2) — Transfer Data Between File Descriptors.