Building a Trillion‑Message Queue: Kafka‑Level Architecture and Implementation

Why Build a Kafka‑Level Message Queue Kernel from Scratch

Standard MQs (Kafka, RocketMQ, Pulsar) satisfy most workloads, but some business scenarios require deep customizations such as bespoke protocols, strict audit/compliance, massive backlog handling, cross‑datacenter DR, and minimal external dependencies. The real challenge is not merely persisting messages but simultaneously achieving:

Million‑TPS write throughput during peaks.

Strong durability despite node failures, network jitter, or rolling upgrades.

Fast read paths that support large consumer groups, catch‑up, replay, and batch consumption.

A complete engineering ecosystem (rate limiting, circuit breaking, monitoring, scaling, migration, recovery).

These four requirements turn the system into a distributed log rather than a simple queue.

Target Goals

Functional Goals

Topic / Partition / Consumer‑Group three‑layer abstraction.

Sequential append writes, batch fetch, offset commit, retry, dead‑letter queue.

Multi‑replica replication, leader election, failover.

Idempotent production and extensible transactional semantics.

Fast lookup by time and offset.

Online scaling, rolling upgrades, historical replay, tiered storage.

Non‑Functional Goals

High throughput: tens of thousands to millions of messages per second per broker.

Low latency: P99 write latency in the millisecond range.

High availability: seconds‑level recovery after broker failure.

Scalability: horizontal scaling of brokers, partitions, and replicas.

Observability: quantifiable end‑to‑end latency, backlog, replication lag, rebalance time, and flush latency.

Core Design Principles

Append‑Only Replicable Log

Write path as sequential as possible.

Read path batch‑oriented.

Minimize metadata changes.

Stream‑oriented inter‑node replication.

These principles stem from three hardware facts:

Disks excel at sequential writes, not random writes.

Networks favor bulk transfers over many small packets.

Frequent leader changes destabilize distributed consistency.

Page Cache Over Hand‑Written Cache

Writes first land in the OS page cache, avoiding a synchronous disk flush per message.

Hot reads hit the page cache directly in kernel space.

Sequential logs naturally align with pre‑read and write‑back strategies.

The engine therefore focuses on page‑cache‑friendly write patterns, controlling dirty‑page ratios, and preventing cold‑data scans from polluting hot caches.

Partition as the Minimal Throughput Unit

Each partition guarantees internal order.

Parallel partitions increase overall throughput.

Replication occurs per partition.

Consumer groups are assigned at the partition level.

Overall Architecture

A full‑stack broker kernel consists of the following modules: Network Layer: long‑lived connections, codec, request framing. API Handler: processes produce, fetch, offset commit, metadata, JoinGroup, Heartbeat, etc. Replica Manager: leader/follower role, append writes, replication progress, ISR management. Log Manager: creates, rolls, deletes, recovers log segments and builds indexes. Group Coordinator: manages consumer‑group membership, rebalance, assignment, session timeout. Raft Controller: maintains consistency of topics, partitions, replica assignments, and broker membership. Observability: metrics, logs, audit trails, tracing, and key‑event subscriptions.

Produce Request Flow

Producer selects a partition based on routing key.

Request passes through the network layer and is decoded.

Broker performs tenant authentication, quota checks, and flow‑control.

Request enters the partition write queue.

Leader partition appends the batch to the active log segment.

Leader updates the Log End Offset (LEO) and notifies followers.

ISR replicas catch up; leader advances the High Watermark (HW).

After satisfying acks=1 or acks=all, a success response is returned.

Consume Request Flow

Consumer sends fetchOffset to start pulling.

Broker uses the index to locate the segment and physical position.

Batch of messages is read from page cache or disk.

Response is sent when minBytes is accumulated or maxWaitMs expires.

Consumer commits the offset after processing.

These two flows account for roughly 80 % of system performance and reliability.

Message Model

Production‑grade batches contain the following fields (batch‑first‑class design): magic: protocol version for gray‑upgrade. attributes: compression, transaction flag, idempotence flag, etc. timestamp: production or event time. topicId: logical topic identifier. partitionId: partition identifier. baseOffset: batch start offset. batchSize: number of records in the batch. producerId: idempotent producer identifier. producerEpoch: producer generation. baseSequence: batch sequence number. headers: business‑specific extensions. key: routing key. value: message payload.

Batch‑first‑class design yields better compression, validation, replication efficiency, and enables zero‑copy transmission.

+-------------------------------+
| Batch Header                  |
|  - magic                     |
|  - flags                     |
|  - crc                       |
|  - baseOffset                |
|  - recordCount               |
|  - compressionType           |
+-------------------------------+
| Record 1                      |
| Record 2                      |
| Record 3                      |
+-------------------------------+

Storage Kernel

Directory Layout per Partition

/data/topics/order_event/partition-0003/
  00000000000000000000.log
  00000000000000000000.index
  00000000000000000000.timeindex
  00000000000000000000.txnindex
  00000000000000838860.log
  00000000000000838860.index
  leader-epoch-checkpoint
  partition-manifest

.log

: stores real message batches. .index: sparse offset‑to‑physical‑position index. .timeindex: enables time‑based lookups. .txnindex: supports transactional batch location and recovery. leader-epoch-checkpoint: used for truncation and replica recovery.

Why Segment the Log

Unlimited single‑file growth makes recovery time explode.

Retention policies usually roll based on time or size, requiring segment deletion.

Sparse indexes need to be associated with local data for hot‑cold tiering.

After a crash, scanning only the last active segment is sufficient.

Sparse Index Value

Index every fixed byte interval (e.g., 4 KB or 8 KB) and binary‑search the index before scanning the local segment. This trades a small amount of extra storage for fast local lookups.

Production‑Grade Log Segment Implementation

public final class LogSegment {
    private final FileChannel logChannel;
    private final OffsetIndex offsetIndex;
    private final TimeIndex timeIndex;
    private final TxnIndex txnIndex;
    private final long baseOffset;
    private volatile long nextOffset;
    private volatile int bytesSinceLastIndexEntry;

    public synchronized AppendResult append(RecordBatch batch) throws IOException {
        validateBatch(batch);
        long batchBaseOffset = nextOffset;
        ByteBuffer encoded = batch.assignBaseOffset(batchBaseOffset).encode();
        int batchSize = encoded.remaining();
        long physicalPosition = logChannel.size();
        while (encoded.hasRemaining()) {
            logChannel.write(encoded);
        }
        if (bytesSinceLastIndexEntry >= offsetIndex.intervalBytes()) {
            offsetIndex.append(batchBaseOffset, physicalPosition);
            timeIndex.maybeAppend(batch.maxTimestamp(), batchBaseOffset);
            bytesSinceLastIndexEntry = 0;
        } else {
            bytesSinceLastIndexEntry += batchSize;
        }
        if (batch.isTransactional()) {
            txnIndex.append(batchBaseOffset, physicalPosition, batchSize);
        }
        nextOffset += batch.recordCount();
        return new AppendResult(batchBaseOffset, nextOffset, physicalPosition, batchSize);
    }

    private void validateBatch(RecordBatch batch) {
        if (batch.recordCount() <= 0) {
            throw new IllegalArgumentException("empty batch");
        }
        if (!batch.checkCrc()) {
            throw new IllegalArgumentException("corrupted batch");
        }
    }
}

The implementation differs from a naive write() by allocating offsets per batch, handling transactional indexes, and delegating flush decisions to a unified flush strategy.

Crash Recovery Process

Load directory listings and segment list.

Sequentially scan the last active segment, validating batch headers and CRC.

On encountering a corrupted or incomplete batch, truncate to the previous valid position.

Rebuild the index for the truncated segment.

Restore high‑watermark and leader epoch from the checkpoint.

The rule is “truncate rather than gamble” – never continue serving after a detected corruption.

Write Path Design

Key Stages

Connection intake and protocol decoding.

Authentication, ACL, and quota checks.

Routing to the target Topic‑Partition.

Batch aggregation.

Enqueue into the partition write queue.

Leader appends to the log.

Follower pulls to catch up.

After the required acknowledgment semantics, the response is returned.

Any poorly designed stage introduces latency spikes.

Thread Separation

Network threads only receive, decode, and perform quick validation.

Business threads handle aggregation and logical processing.

Partition writer threads perform ordered persistence.

Partition Serial, Global Parallel

Writes within the same partition are serialized; writes across different partitions proceed in parallel. This is realized by a PartitionAppender that consumes a blocking queue of AppendTask objects, merges them into a single MemoryRecords batch, and invokes log.appendAsLeader:

public final class PartitionAppender implements Runnable {
    private BlockingQueue<AppendTask> queue;
    private Log log;
    private ReplicaTracker replicaTracker;

    @Override
    public void run() {
        while (!Thread.currentThread().isInterrupted()) {
            AppendTask task = queue.take();
            List<AppendTask> batch = drainBatch(task);
            MemoryRecords records = MemoryRecords.merge(batch);
            AppendInfo appendInfo = log.appendAsLeader(records);
            replicaTracker.onLeaderAppend(appendInfo);
            for (AppendTask item : batch) {
                item.complete(appendInfo.baseOffset(), appendInfo.lastOffset());
            }
        }
    }
}

Benefits include per‑partition order, reduced system‑call frequency via batch merging, and isolated hot partitions for targeted governance.

Flush Strategies

async

: write to page cache and return immediately; a background thread flushes to disk – highest throughput. flush‑interval: trigger flush after a fixed time or byte threshold – balances throughput and durability. sync: flush after each batch before acknowledging – strongest durability but highest latency.

Reliability also depends on multi‑replica replication and whether the acknowledgment requires the entire ISR to be caught up.

Read Path Design

Batch + Long‑Polling

minBytes

defines the minimum accumulated bytes before a response. maxWaitMs caps the wait time even if data is insufficient.

This balances throughput against latency.

Read Flow

Locate the segment using fetchOffset.

Binary‑search the .index to find the physical position.

Sequentially scan batches until maxBytes is reached.

Decompress if needed or send compressed batches directly.

Use zero‑copy (e.g., Netty FileRegion) or aggregation to return data.

Zero‑Copy Send Example

public final class FetchSender {
    public void send(FileChannel channel,
                     long position,
                     long count,
                     Channel nettyChannel,
                     CompletableFuture<Void> future) {
        FileRegion region = new DefaultFileRegion(channel, position, count);
        nettyChannel.writeAndFlush(region).addListener(result -> {
            if (result.isSuccess()) {
                future.complete(null);
            } else {
                future.completeExceptionally(result.cause());
            }
        });
    }
}

Practical considerations include fragmenting large files, handling TLS (which disables true zero‑copy), and falling back to full decode when the consumer requests uncompressed messages.

Time‑Based Replay

Without a time index, replaying a window such as “yesterday 10:00‑10:30” would be painful. Time indexes enable fast back‑track for debugging, compensation, and audit.

Replication and Consistency

Key Offsets

LEO

(Log End Offset): the end of the local log. HW (High Watermark): the highest offset that is safely replicated to all ISR members and therefore visible to consumers. LSO (Last Stable Offset): used in transactional scenarios as the last committed offset.

Consumers read only up to HW (or LSO for transactions), never raw LEO, to avoid serving data that may be truncated after a leader failure.

ISR Mechanism

A replica stays in the In‑Sync Replica set if replication lag and offset lag stay below configured thresholds and the replica remains alive with heartbeats. acks=all succeeds only when the required number of ISR replicas have replicated the batch.

Replication Flow

Follower sends its current fetchOffset to the leader.

Leader returns new batches.

Follower appends to its local log and updates its LEO.

Leader, upon receiving progress from ISR members, advances HW.

This pull‑based approach reuses the consumer protocol for replica sync and simplifies flow control, especially for cross‑datacenter links.

Leader Election via Raft

Topic creation, partition changes, and broker membership are logged in a Raft‑based metadata log.

Raft leader decides control‑plane decisions.

Each broker caches the latest snapshot and incremental events.

Data‑plane replication and control‑plane consensus are deliberately decoupled: the control plane manages leader identity, ISR membership, and partition placement, while the data plane handles append, replication, and truncation.

Replica Management Example

public final class PartitionReplica {
    private volatile long leaderEpoch;
    private volatile long highWatermark;
    private final ConcurrentMap<Integer, ReplicaState> followers = new ConcurrentHashMap<>();

    public void updateFollowerProgress(int brokerId, long leo, long fetchTimeMs) {
        followers.compute(brokerId, (id, state) -> {
            ReplicaState next = state == null ? new ReplicaState() : state;
            next.leo = leo;
            next.lastFetchTimeMs = fetchTimeMs;
            return next;
        });
        maybeAdvanceHighWatermark();
    }

    private void maybeAdvanceHighWatermark() {
        long candidate = followers.values().stream()
            .mapToLong(s -> s.leo)
            .min()
            .orElse(highWatermark);
        if (candidate > highWatermark) {
            highWatermark = candidate;
        }
    }
}

Real implementations also handle leader LEO, ISR expansion/reduction, slow‑replica eviction, epoch changes, and log truncation.

Metadata System

Metadata stores more than topic names: broker registration, heartbeats, tenant quotas, ACLs, consumer‑group state, and partition‑replica assignments. Embedding Raft offers deployment simplicity, linearizable metadata operations, and fast snapshot‑based recovery, while keeping high‑frequency data‑plane state out of the Raft log.

Consumer‑Group Coordination

State Machine

Empty → PreparingRebalance → CompletingRebalance → Stable → Dead

Key Requests

JoinGroup

: member joins and reports supported assignor. SyncGroup: group leader distributes final assignment. Heartbeat: keeps session alive. LeaveGroup: member voluntarily leaves. OffsetCommit: commits consumption progress.

Rebalance Storm Causes

Session timeout too short → accidental member eviction.

Bulk consumer restarts → frequent group changes.

Complex assignor algorithm → long rebalance latency.

Full‑group migration on every member change.

Mitigation Example

public void onMemberJoin(Group group, Member member) {
    group.add(member);
    if (group.isStable()) {
        group.transitionToPreparingRebalance();
    }
    rebalanceScheduler.schedule(group.groupId(), group.rebalanceDelayMs());
}

Key is rigorous state transitions, traceable timeouts, and recoverable error paths.

Engineering Upgrades for High Concurrency, Scalability, and Stability

High‑Concurrency Practices

Batch producers using batch.size and linger.ms.

Broker merges same‑partition write requests before persisting.

Followers pull replication in batches.

Memory Pooling

Network layer reuses ByteBuf pools.

Codec buffers are recycled to avoid short‑lived objects.

Large batch objects stay out of the old generation, reducing Full GC risk.

Multi‑Level Queues

Ingress queue shields the network layer.

Partition write queue protects the log layer.

Replication queue protects follower sync.

Hot‑Topic Isolation

Dedicated thread pools or brokers for hot topics.

Tenant‑level token‑bucket quotas prevent a single large customer from overwhelming the system.

Flow Control & Back‑Pressure

Connection‑level rate limiting.

Topic‑level write quotas.

Tenant‑level token buckets.

Broker‑level disk‑watermark protection.

Partition‑write‑queue length thresholds for circuit breaking.

A layered approach applies coarse rejection at the network layer, quota checks at the API layer, and disk‑delay protection at the storage layer.

Scalability Design

Data expansion: add brokers, increase partitions, redistribute replicas.

Capacity expansion: add disks, tiered storage.

Governance expansion: metadata, flow‑control, monitoring scale with the system.

Observability

Production/consumption/replication TPS.

P50/P95/P99 write and fetch latency.

Page‑cache dirty pages, flush latency, disk queue depth.

Per‑partition LEO, HW, replication lag.

Consumer‑group lag, rebalance count, heartbeat timeouts.

Request rejection rate, limit‑hit rate, dead‑letter count.

Key events: leader switches, ISR changes, segment roll, log truncation, consumer‑group rebalances.

Advanced Production‑Grade Features

Idempotent Production

Assign a unique producerId per producer.

Maintain (producerId, producerEpoch, sequence) state per partition.

On retry, duplicate sequence numbers are treated as idempotent replays.

Transactional Messaging

Open a transaction, write to multiple partitions, then commit or abort.

Consumers read only up to the Last Stable Offset (LSO) for committed data.

Dead‑Letter and Retry

In‑partition retry.

Delayed‑retry topics.

Dead‑letter queue for permanently failed messages, storing failure reason, retry count, original topic, and offset.

Delayed Messages

Time‑wheel with expiration delivery.

Delayed topic tiered storage that re‑routes messages after the delay expires.

First‑stage implementations usually adopt a “tiered delay queue” to keep complexity manageable.

Real‑World Business Case: Order Center Supporting Peak Traffic

Scenario: an e‑commerce order flow (order → inventory reservation → payment → invoice → logistics → completion) requires strict ordering per order, high throughput, and the ability to replay windows for compensation.

Topic: order_event Partition key: orderId (ensures the same order stays in one partition).

Headers: tenantId, traceId, eventType (for audit and quota isolation).

Benefits: natural ordering per order, parallelism across orders, and tenant‑level audit/quota.

Failure Scenarios

Payment service avalanche: consumer lag spikes, broker throttles hot order topics, protects core order writes, and later catches up.

Leader crash: Raft detects broker down, selects new leader from ISR, producers refresh metadata, consumers continue from HW.

Mis‑consumption requiring replay: time index locates the 10:00‑10:15 window, a temporary consumer group pulls from that point, and compensates erroneous orders.

Kubernetes & Cloud‑Native Deployment

StatefulSet as the First Choice

Stable pod names map directly to broker.id.

Each pod gets its own PVC for log storage.

Rolling upgrades are controlled.

Key Deployment Practices

Use high‑performance SSD or local NVMe for data disks.

Pod anti‑affinity to avoid placing primary and replica on the same node.

PodDisruptionBudget to prevent simultaneous broker eviction.

Separate liveness (process started) and readiness (metadata ready) probes.

PreStop hook migrates the leader before pod termination.

Storage & Network Recommendations

Separate data and log directories.

Different ports for control‑plane and data‑plane traffic.

Cross‑region replication uses dedicated bandwidth and compression.

Cold historical data is gradually off‑loaded to object storage.

Implementation Roadmap: From 0 to 1

Single‑Node Prototype : implement message model, batch format, sequential log, offset semantics, local file persistence, basic fetch and replay, crash recovery.

High‑Performance Single Node : add batch writes, sparse indexes, zero‑copy reads, long‑polling, memory pooling, dedicated flush thread.

Distributed Replication : leader/follower, ISR, HW, Raft‑based metadata control.

Consumer Groups & Governance : group coordination, rebalance, flow control, metrics, alerts.

Advanced Features : idempotent production, transactions, dead‑letter, delayed messages, tiered storage.

Each stage has clear acceptance criteria (throughput, latency, recovery, stability) to avoid building an overly‑big system prematurely.

Common Pitfalls Checklist

Over‑frequent index updates cause index bloat and write amplification.

Treating High Watermark as the log end leads to consumers reading unstable data.

Global lock in rebalance logic stalls the whole coordinator on a single member glitch.

Ignoring real disk behavior; cloud‑disk jitter can explode latency if not accounted for.

Skipping crash‑recovery drills – without power‑off, crash, half‑write, follower lag, and leader flip tests the system is not production‑ready.

Conclusion

Building a Kafka‑grade message‑queue kernel from scratch is not about writing a simple broker process; it requires mastering sequential logs, batch protocols, ISR and high‑watermark semantics, consumer‑group coordination, flow control, observability, and a staged evolution from prototype to production. When these capabilities are understood and implemented, you gain a powerful foundation for high‑throughput, highly reliable distributed infrastructure.