Build a Message Streaming Engine
Build a Kafka-inspired message broker from scratch using Python.
Introduction
30 Second Summary
Every time you get a real-time notification, see a live dashboard update, or watch an activity feed refresh, there's a message streaming system routing data between services behind the scenes. Apache Kafka is the industry standard for this, processing trillions of messages per day.
In this project, you will build MiniKafka from scratch using only Python's standard library. You'll implement the same core architecture that makes Kafka fast: an append-only segmented log, a binary wire protocol, and consumer-controlled offsets.
What You'll Build
A single-broker message streaming system with segmented storage, a custom binary protocol over TCP, and producer/consumer clients that demonstrate partition routing and offset-based replay.
By the end of this project, you'll have:
- A storage engine with append-only segmented commit logs, offset tracking, and CRC32 integrity validation.
- A TCP message broker that accepts produce and fetch requests over a custom binary wire protocol with length-prefixed framing.
- Producer and consumer clients with key-based partition routing, offset-based replay, and end-to-end message streaming.
- Secret Mission: Add consumer group coordination with range-based partition assignment and rebalancing.
Are there any prerequisites?
This project assumes you're comfortable writing Python and familiar with basic file I/O. You should also have some exposure to TCP sockets and binary data (the struct module). If you've ever packed or unpacked bytes in Python, you're ready.
Set Up the Project Structure
MiniKafka has several distinct components that work together: a storage engine, a wire protocol, a broker, and clients. Each component lives in its own Python module, keeping responsibilities cleanly separated.
In this step, you'll create the project directory with all five module files and confirm your Python version supports the syntax used throughout the project.
In this step, get ready to:
- Create the minikafka/ project directory with five Python module files.
- Verify Python 3.10+ is installed.
- Open the project in Cursor.
Create the project directory and module files
MiniKafka's architecture splits into five files. storage.py handles the commit log, protocol.py defines the binary wire format, broker.py runs the TCP server, client.py provides producer and consumer classes, and demo.py ties everything together.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Terminal and press Enter to open it.
- Move to your Desktop by running this command:
cd ~/Desktop- Create the project directory and move into it by running these commands:
mkdir minikafka
cd minikafka- Create the five empty module files by running this command:
touch storage.py protocol.py broker.py client.py demo.py- Confirm the files exist by running:
lsYou should see storage.py, protocol.py, broker.py, client.py, and demo.py listed in the output.
Don't see all five files?
Make sure you ran the touch command from inside the minikafka directory. Run pwd to check your current location.
Help me troubleshoot this
Verify Python 3.10+ is available
This project uses Python 3.10+ features like union type syntax (bytes | None) and match statements. Without the right version, you'll hit syntax errors on the first import.
- Check your Python version by running:
python3 --versionYou should see Python 3.10 or higher (e.g. Python 3.12.4).
Seeing a version lower than 3.10?
If your system Python reports 3.9 or lower, you need to install a newer version. On macOS, the system-bundled Python is often outdated. Install Python 3.10+ from python.org or use a version manager like pyenv.
Help me install the right Python version
Open the project in Cursor
With your files in place, open the project folder in Cursor so you have the editor and integrated terminal ready for the next steps.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Cursor and press Enter to open it.
- In Cursor, click File in the top menu bar.
- Click Open Folder.
- Navigate to Desktop, select the minikafka folder, then click Open.
You should see all five .py files listed in the Explorer sidebar on the left.
Why five separate files?
Each file maps to one layer of the system. This mirrors how real distributed systems are built: the storage layer knows nothing about the network, and the protocol layer knows nothing about how data is persisted.
Keeping them separate means you can test each layer in isolation. You'll see this pay off when you run storage.py directly to validate the commit log before wiring up the broker.
Your project structure is ready. Next up, you'll build the storage engine and implement the append-only commit log that sits at the heart of every message streaming system.
Build the Storage Engine
The heart of Apache Kafka is its storage layer: the commit log. Every message that arrives at the broker gets appended to the end of a log file. There are no random writes, no updates, no deletes. This append-only design is what gives Kafka O(1) write performance regardless of data volume.
In this step, you'll build the commit log from scratch. You'll implement a binary record format with integrity checks, segment files that cap at a fixed size, and a manager that handles rotation and crash recovery.
In this step, get ready to:
- Implement the Record class with binary serialization and CRC32 integrity validation.
- Build a Segment class that wraps individual log files on disk.
- Build the CommitLog manager with segment rotation and startup recovery.
Implement Record serialization
Every message stored in the log needs a consistent binary format. The format includes an offset (the record's position in the log), a CRC32 checksum for corruption detection, and the key-value payload.
The on-disk layout per record is: [4B record_size][8B offset][4B crc32][4B key_len][key_bytes][4B value_len][value_bytes]. All integers use big-endian byte order.
- Open storage.py in Cursor and add the following imports, along with the Record class:
"""Append-only commit log with segmented storage."""
import os
import struct
import zlib
from pathlib import Path
from threading import Lock
class Record:
"""A single record in the commit log."""
def __init__(self, offset: int, key: bytes | None, value: bytes):
self.offset = offset
self.key = key
self.value = value
def serialize(self) -> bytes:
# Convert key to bytes, handle null keys with key_len = -1
key_bytes = self.key if self.key is not None else b""
key_len = len(self.key) if self.key is not None else -1
# CRC covers key + value bytes for integrity validation
crc = zlib.crc32(key_bytes + self.value) & 0xFFFFFFFF
# Pack offset (8B), crc (4B), key_len (4B signed)
body = struct.pack(">qIi", self.offset, crc, key_len)
if key_len >= 0:
body += self.key
body += struct.pack(">I", len(self.value))
body += self.value
# Prefix with total body length for framing
return struct.pack(">I", len(body)) + body
@classmethod
def deserialize(cls, data: bytes, pos: int) -> tuple["Record", int]:
# Read body length to know where this record ends
body_len = struct.unpack_from(">I", data, pos)[0]
pos += 4
end = pos + body_len
# Unpack fixed header fields
offset, crc, key_len = struct.unpack_from(">qIi", data, pos)
pos += 16
# Read key bytes (null if key_len == -1)
key = None
if key_len >= 0:
key = data[pos : pos + key_len]
pos += key_len
# Read value bytes
value_len = struct.unpack_from(">I", data, pos)[0]
pos += 4
value = data[pos : pos + value_len]
pos += value_len
# Validate CRC to detect disk corruption
key_bytes = key if key is not None else b""
expected_crc = zlib.crc32(key_bytes + value) & 0xFFFFFFFF
if crc != expected_crc:
raise ValueError(f"CRC mismatch at offset {offset}")
return cls(offset, key, value), endWhat does this code do?
The Record class handles two jobs: turning a message into bytes for disk storage, and reconstructing a message from raw bytes.
- serialize() packs the offset, CRC32, key length, key, value length, and value into a big-endian byte sequence. The 4-byte record size prefix lets the reader know exactly how many bytes to consume.
- deserialize() reverses the process. It reads the size prefix, unpacks the header, extracts key and value, then validates the CRC32 checksum. If corruption is detected, it raises an error.
- Null keys use key_len = -1 as a sentinel. The CRC still covers an empty key in this case so the checksum stays consistent.
Why big-endian byte order?
Big-endian (network byte order) is the standard for binary protocols because it's unambiguous across machines. Kafka's wire protocol uses the same convention. The > prefix in struct.pack tells Python to use big-endian.
- Save storage.py.
You won't see output yet. The next chunk builds the segment layer that uses these records.
Seeing a syntax error on the | in bytes | None?
The union type syntax bytes | None requires Python 3.10 or higher. Run python3 --version to confirm your version. If you're on an older version, revisit Step 1 to upgrade.
help me fix a Python version error with union type syntax
Implement Segment and CommitLog
Kafka doesn't write to one infinitely growing file. It splits the log into fixed-size segments. When a segment reaches its size limit, a new one is created. This makes cleanup and retention simple because you can delete entire segment files.
- Add the Segment class below the Record class in storage.py:
class Segment:
"""A single log segment file."""
def __init__(self, log_dir: Path, base_offset: int):
self.base_offset = base_offset
# Segment file named by its starting offset, zero-padded to 20 digits
self.path = log_dir / f"{base_offset:020d}.log"
self.path.touch(exist_ok=True)
self._size = self.path.stat().st_size
@property
def size(self) -> int:
return self._size
def append(self, record: Record) -> None:
# Serialize and append in binary mode
data = record.serialize()
with open(self.path, "ab") as f:
f.write(data)
self._size += len(data)
def read_from(self, start_offset: int, max_bytes: int) -> list[Record]:
records = []
with open(self.path, "rb") as f:
data = f.read()
pos = 0
bytes_collected = 0
# Scan forward through all records in this segment
while pos < len(data) and bytes_collected < max_bytes:
try:
record, new_pos = Record.deserialize(data, pos)
except (struct.error, ValueError, IndexError):
break
# Only collect records at or after the requested offset
if record.offset >= start_offset:
records.append(record)
bytes_collected += (new_pos - pos)
pos = new_pos
return recordsWhat does this code do?
- Each segment file is named by its base offset, zero-padded to 20 digits (e.g., 00000000000000000000.log). This matches Kafka's naming convention and keeps files sorted lexicographically.
- append() opens the file in binary-append mode. Writes always go to the end of the file.
- read_from() scans forward through the segment, deserializing records and collecting those at or after the requested offset. The max_bytes cap prevents unbounded reads.
Now add the CommitLog class below Segment. This is the top-level manager that ties segments together.
- Add the CommitLog class at the bottom of storage.py (below the Segment class):
class CommitLog:
"""Manages a partitioned, segmented append-only log."""
def __init__(self, data_dir: Path, topic: str, partition: int, max_segment_bytes: int = 1_048_576):
# Each topic-partition gets its own directory
self.log_dir = data_dir / f"{topic}-{partition}"
self.log_dir.mkdir(parents=True, exist_ok=True)
self.max_segment_bytes = max_segment_bytes
self._lock = Lock()
self._next_offset = 0
self._segments: list[Segment] = []
self._recover()
def _recover(self) -> None:
"""Rebuild state from existing segment files on startup."""
log_files = sorted(self.log_dir.glob("*.log"))
if not log_files:
self._segments.append(Segment(self.log_dir, 0))
return
for log_file in log_files:
base_offset = int(log_file.stem)
self._segments.append(Segment(self.log_dir, base_offset))
# Scan last segment to find the highest offset
last_segment = self._segments[-1]
with open(last_segment.path, "rb") as f:
data = f.read()
pos = 0
while pos < len(data):
try:
record, pos = Record.deserialize(data, pos)
self._next_offset = record.offset + 1
except (struct.error, ValueError, IndexError):
break
def append(self, key: bytes | None, value: bytes) -> int:
"""Append a record and return its assigned offset."""
with self._lock:
offset = self._next_offset
self._next_offset += 1
# Rotate to a new segment if the active one is full
active = self._segments[-1]
if active.size >= self.max_segment_bytes:
active = Segment(self.log_dir, offset)
self._segments.append(active)
record = Record(offset, key, value)
active.append(record)
return offset
def read(self, start_offset: int, max_bytes: int = 65536) -> list[Record]:
"""Read records starting from the given offset."""
with self._lock:
# Find the segment that contains start_offset
target = self._segments[0]
for seg in self._segments:
if seg.base_offset <= start_offset:
target = seg
else:
break
return target.read_from(start_offset, max_bytes)What does this code do?
- CommitLog manages a list of segments for one topic-partition. Each topic-partition gets its own directory (e.g., user-events-0/).
- _recover() runs on startup. It scans existing .log files, rebuilds the segment list, and finds the highest offset so new appends continue from the right position. This is crash recovery.
- append() assigns the next offset, checks if the active segment is full, rotates if needed, then writes. The lock ensures thread safety.
- read() locates the correct segment by comparing base offsets, then delegates to read_from().
Why is the broker stateless about consumer progress?
Notice that CommitLog tracks its own write position (_next_offset) but has no idea where any consumer is reading from. The consumer passes its desired offset in each read call.
This is Kafka's fundamental architectural insight. By pushing offset tracking to the consumer, the broker stays simple and scalable. Consumers can replay old messages just by seeking to an earlier offset.
- Save storage.py.
You won't see visible output yet. The next substep adds a test block to validate the engine.
Seeing an indentation error?
Make sure the Segment and CommitLog classes are at the module level (no indentation on the class keyword). Nested classes would cause NameError when one tries to reference the other.
help me debug an indentation or NameError in storage.py
Validate your storage engine
The design decisions embedded in this code mirror real Kafka behavior. Serial appends always go to the active (last) segment. New segments are created with a base offset equal to the first record they'll contain. Reads locate the correct segment by comparing base offsets, then scan forward.
Let's add a test that proves all of this works, including segment rotation.
- Add the following __main__ block at the very bottom of storage.py (below the CommitLog class, at module level):
if __name__ == "__main__":
import tempfile
with tempfile.TemporaryDirectory() as tmp:
# Use 256 bytes max so we can observe segment rotation
log = CommitLog(Path(tmp), "test-topic", 0, max_segment_bytes=256)
for i in range(10):
off = log.append(f"key-{i}".encode(), f"value-{i}".encode())
print(f"Appended at offset {off}")
print("\nReading from offset 0:")
for r in log.read(0):
print(f" offset={r.offset} key={r.key} value={r.value}")
print("\nReading from offset 5:")
for r in log.read(5):
print(f" offset={r.offset} key={r.key} value={r.value}")What does this test do?
- It creates a CommitLog with a tiny 256-byte segment limit so rotation happens quickly during the test.
- It appends 10 records, each printing their assigned offset (0 through 9).
- It reads all records from offset 0, then reads again from offset 5 to demonstrate that consumers can start reading from any position.
- Save storage.py.
- Run the test by executing this command in your terminal:
python3 storage.pyYou should see 10 lines showing offsets 0 through 9, followed by all records read from offset 0, then records 5 through 9 read from offset 5.
Seeing a ModuleNotFoundError or FileNotFoundError?
Make sure your terminal is inside the minikafka/ directory when you run the command. The tempfile module is part of the standard library, so it should always be available.
If you see a NameError for Record or Segment, check that all three classes are defined at the top level of the file (not nested inside another class or function).
help me debug an error when running python3 storage.py
✔️ Awesome, I've got everything!
Great. Double check you've saved storage.py and confirmed the test output matches what's described above.
ⓧ I'd like to double check the full code
Here's the complete storage.py file. Compare yours against it:
"""Append-only commit log with segmented storage."""
import os
import struct
import zlib
from pathlib import Path
from threading import Lock
class Record:
"""A single record in the commit log."""
def __init__(self, offset: int, key: bytes | None, value: bytes):
self.offset = offset
self.key = key
self.value = value
def serialize(self) -> bytes:
key_bytes = self.key if self.key is not None else b""
key_len = len(self.key) if self.key is not None else -1
crc = zlib.crc32(key_bytes + self.value) & 0xFFFFFFFF
body = struct.pack(">qIi", self.offset, crc, key_len)
if key_len >= 0:
body += self.key
body += struct.pack(">I", len(self.value))
body += self.value
return struct.pack(">I", len(body)) + body
@classmethod
def deserialize(cls, data: bytes, pos: int) -> tuple["Record", int]:
body_len = struct.unpack_from(">I", data, pos)[0]
pos += 4
end = pos + body_len
offset, crc, key_len = struct.unpack_from(">qIi", data, pos)
pos += 16
key = None
if key_len >= 0:
key = data[pos : pos + key_len]
pos += key_len
value_len = struct.unpack_from(">I", data, pos)[0]
pos += 4
value = data[pos : pos + value_len]
pos += value_len
key_bytes = key if key is not None else b""
expected_crc = zlib.crc32(key_bytes + value) & 0xFFFFFFFF
if crc != expected_crc:
raise ValueError(f"CRC mismatch at offset {offset}")
return cls(offset, key, value), end
class Segment:
"""A single log segment file."""
def __init__(self, log_dir: Path, base_offset: int):
self.base_offset = base_offset
self.path = log_dir / f"{base_offset:020d}.log"
self.path.touch(exist_ok=True)
self._size = self.path.stat().st_size
@property
def size(self) -> int:
return self._size
def append(self, record: Record) -> None:
data = record.serialize()
with open(self.path, "ab") as f:
f.write(data)
self._size += len(data)
def read_from(self, start_offset: int, max_bytes: int) -> list[Record]:
records = []
with open(self.path, "rb") as f:
data = f.read()
pos = 0
bytes_collected = 0
while pos < len(data) and bytes_collected < max_bytes:
try:
record, new_pos = Record.deserialize(data, pos)
except (struct.error, ValueError, IndexError):
break
if record.offset >= start_offset:
records.append(record)
bytes_collected += (new_pos - pos)
pos = new_pos
return records
class CommitLog:
"""Manages a partitioned, segmented append-only log."""
def __init__(self, data_dir: Path, topic: str, partition: int, max_segment_bytes: int = 1_048_576):
self.log_dir = data_dir / f"{topic}-{partition}"
self.log_dir.mkdir(parents=True, exist_ok=True)
self.max_segment_bytes = max_segment_bytes
self._lock = Lock()
self._next_offset = 0
self._segments: list[Segment] = []
self._recover()
def _recover(self) -> None:
log_files = sorted(self.log_dir.glob("*.log"))
if not log_files:
self._segments.append(Segment(self.log_dir, 0))
return
for log_file in log_files:
base_offset = int(log_file.stem)
self._segments.append(Segment(self.log_dir, base_offset))
last_segment = self._segments[-1]
with open(last_segment.path, "rb") as f:
data = f.read()
pos = 0
while pos < len(data):
try:
record, pos = Record.deserialize(data, pos)
self._next_offset = record.offset + 1
except (struct.error, ValueError, IndexError):
break
def append(self, key: bytes | None, value: bytes) -> int:
with self._lock:
offset = self._next_offset
self._next_offset += 1
active = self._segments[-1]
if active.size >= self.max_segment_bytes:
active = Segment(self.log_dir, offset)
self._segments.append(active)
record = Record(offset, key, value)
active.append(record)
return offset
def read(self, start_offset: int, max_bytes: int = 65536) -> list[Record]:
with self._lock:
target = self._segments[0]
for seg in self._segments:
if seg.base_offset <= start_offset:
target = seg
else:
break
return target.read_from(start_offset, max_bytes)
if __name__ == "__main__":
import tempfile
with tempfile.TemporaryDirectory() as tmp:
log = CommitLog(Path(tmp), "test-topic", 0, max_segment_bytes=256)
for i in range(10):
off = log.append(f"key-{i}".encode(), f"value-{i}".encode())
print(f"Appended at offset {off}")
print("\nReading from offset 0:")
for r in log.read(0):
print(f" offset={r.offset} key={r.key} value={r.value}")
print("\nReading from offset 5:")
for r in log.read(5):
print(f" offset={r.offset} key={r.key} value={r.value}")Your storage engine is complete. You now have an append-only commit log that handles binary serialization, CRC32 integrity checks, segment rotation, and crash recovery. Next up, you'll design the binary wire protocol that lets clients talk to the broker over TCP.
Design the Wire Protocol
Your storage engine can persist records to disk. Now you need a way for clients to talk to the broker over the network.
Real Kafka uses a binary wire protocol over TCP where every message is size-delimited. In this step, you will implement that same pattern: length-prefixed framing with a one-byte API key that tells the broker how to interpret the payload.
In this step, get ready to:
- Define the binary frame format with dataclass message types.
- Implement length-prefixed framing, encoders, and decoders.
- Verify the protocol with a round-trip encode/decode test.
Define the frame format
Every message on the wire follows a fixed structure: a 4-byte length prefix, a 1-byte API key identifying the request type, then the payload bytes. This lets the receiver know exactly how many bytes to read before parsing.
- Open protocol.py in Cursor and add the protocol type definitions:
"""Binary wire protocol for MiniKafka.
Frame format: [4B length][1B api_key][payload...]
All integers are big-endian.
"""
import struct
import socket
from dataclasses import dataclass, field
# API key constants matching Kafka's numbering
API_PRODUCE = 0
API_FETCH = 1
@dataclass
class ProduceRequest:
topic: str
partition: int
key: bytes | None
value: bytes
@dataclass
class FetchRequest:
topic: str
partition: int
offset: int
max_bytes: int
@dataclass
class ProduceResponse:
offset: int
error: int = 0
@dataclass
class RecordData:
offset: int
key: bytes | None
value: bytes
@dataclass
class FetchResponse:
records: list[RecordData] = field(default_factory=list)
error: int = 0What does this code do?
This establishes the vocabulary of your protocol. API_PRODUCE = 0 and API_FETCH = 1 mirror Kafka's actual API key numbering.
Each dataclass represents one message type. ProduceRequest carries a record to store. FetchRequest asks for records starting at a given offset. The responses carry back the result.
Why a binary protocol instead of HTTP/JSON?
Kafka's wire protocol is binary and length-prefixed specifically for performance. It enables zero-copy transfer via sendfile() and efficient batching. The broker never interprets message content. It just stores and forwards bytes.
Using HTTP/JSON would hide the very design decisions that make Kafka fast. Our protocol mirrors this binary-over-TCP pattern so you understand why it matters.
You will not see a change yet. The next chunk adds the functions that encode and decode these types.
Implement the framing helpers and encode/decode functions
With your message types defined, you need functions to serialize them into bytes and read them off a TCP socket. The framing layer handles length-prefixed reads so the receiver always knows how many bytes to expect.
- Add the framing helpers below the dataclass definitions in protocol.py:
def recv_exact(sock: socket.socket, n: int) -> bytes:
"""Read exactly n bytes from socket."""
buf = bytearray()
while len(buf) < n:
chunk = sock.recv(n - len(buf))
if not chunk:
raise ConnectionError("Connection closed")
buf.extend(chunk)
return bytes(buf)
def recv_frame(sock: socket.socket) -> bytes:
"""Read a length-prefixed frame."""
length_bytes = recv_exact(sock, 4)
length = struct.unpack(">I", length_bytes)[0]
return recv_exact(sock, length)
def send_frame(sock: socket.socket, payload: bytes) -> None:
"""Send a length-prefixed frame."""
sock.sendall(struct.pack(">I", len(payload)) + payload)What does this code do?
- recv_exact loops until it has read exactly n bytes from the socket. TCP can deliver data in chunks smaller than what you asked for, so this loop is essential.
- recv_frame reads the 4-byte length prefix, unpacks it as a big-endian unsigned int, then reads exactly that many payload bytes.
- send_frame prepends the 4-byte length and sends everything in one call.
- Add the request encoder functions below the framing helpers:
def encode_produce_request(req: ProduceRequest) -> bytes:
topic_bytes = req.topic.encode()
key_bytes = req.key if req.key is not None else b""
key_len = len(req.key) if req.key is not None else -1
buf = struct.pack(">B", API_PRODUCE)
buf += struct.pack(">H", len(topic_bytes)) + topic_bytes
buf += struct.pack(">I", req.partition)
buf += struct.pack(">i", key_len)
if key_len >= 0:
buf += key_bytes
buf += struct.pack(">I", len(req.value)) + req.value
return buf
def encode_fetch_request(req: FetchRequest) -> bytes:
topic_bytes = req.topic.encode()
buf = struct.pack(">B", API_FETCH)
buf += struct.pack(">H", len(topic_bytes)) + topic_bytes
buf += struct.pack(">I", req.partition)
buf += struct.pack(">q", req.offset)
buf += struct.pack(">I", req.max_bytes)
return bufWhat does this code do?
Each encoder packs fields into bytes following the layout defined in the module docstring. The first byte is always the API key so the broker knows which decoder to use.
- Produce payload: [1B api_key][2B topic_len][topic][4B partition][4B key_len][key][4B value_len][value]. Null keys use key_len = -1 (signed int).
- Fetch payload: [1B api_key][2B topic_len][topic][4B partition][8B offset][4B max_bytes].
- Add the response encoder functions below the request encoders:
def encode_produce_response(resp: ProduceResponse) -> bytes:
return struct.pack(">Bq", resp.error, resp.offset)
def encode_fetch_response(resp: FetchResponse) -> bytes:
buf = struct.pack(">BI", resp.error, len(resp.records))
for rec in resp.records:
key_bytes = rec.key if rec.key is not None else b""
key_len = len(rec.key) if rec.key is not None else -1
buf += struct.pack(">q", rec.offset)
buf += struct.pack(">i", key_len)
if key_len >= 0:
buf += key_bytes
buf += struct.pack(">I", len(rec.value)) + rec.value
return bufWhat does this code do?
- encode_produce_response packs a 1-byte error code and an 8-byte offset. That is all the producer needs back.
- encode_fetch_response packs the error code, a record count, then each record's offset, key, and value. The consumer uses this to rebuild the messages.
- Add the decoder functions below the response encoders:
def decode_request(data: bytes) -> ProduceRequest | FetchRequest:
pos = 0
api_key = struct.unpack_from(">B", data, pos)[0]
pos += 1
topic_len = struct.unpack_from(">H", data, pos)[0]
pos += 2
topic = data[pos : pos + topic_len].decode()
pos += topic_len
partition = struct.unpack_from(">I", data, pos)[0]
pos += 4
if api_key == API_PRODUCE:
key_len = struct.unpack_from(">i", data, pos)[0]
pos += 4
key = None
if key_len >= 0:
key = data[pos : pos + key_len]
pos += key_len
value_len = struct.unpack_from(">I", data, pos)[0]
pos += 4
value = data[pos : pos + value_len]
return ProduceRequest(topic, partition, key, value)
elif api_key == API_FETCH:
offset = struct.unpack_from(">q", data, pos)[0]
pos += 8
max_bytes = struct.unpack_from(">I", data, pos)[0]
return FetchRequest(topic, partition, offset, max_bytes)
raise ValueError(f"Unknown API key: {api_key}")
def decode_produce_response(data: bytes) -> ProduceResponse:
error, offset = struct.unpack(">Bq", data)
return ProduceResponse(offset=offset, error=error)
def decode_fetch_response(data: bytes) -> FetchResponse:
pos = 0
error, num_records = struct.unpack_from(">BI", data, pos)
pos += 5
records = []
for _ in range(num_records):
offset = struct.unpack_from(">q", data, pos)[0]
pos += 8
key_len = struct.unpack_from(">i", data, pos)[0]
pos += 4
key = None
if key_len >= 0:
key = data[pos : pos + key_len]
pos += key_len
value_len = struct.unpack_from(">I", data, pos)[0]
pos += 4
value = data[pos : pos + value_len]
pos += value_len
records.append(RecordData(offset=offset, key=key, value=value))
return FetchResponse(records=records, error=error)What does this code do?
- decode_request reads the first byte to determine the API key, then branches into Produce or Fetch decoding. This is the broker-side decoder.
- decode_produce_response and decode_fetch_response are the client-side decoders that unpack what the broker sends back.
- The pos variable tracks position as each field is unpacked sequentially. struct.unpack_from reads from a specific offset within the buffer without slicing.
- Save protocol.py.
Seeing a syntax error on save?
- Check that you are running Python 3.10 or higher. The ProduceRequest | FetchRequest union syntax requires 3.10+.
- Verify all the format strings use the correct specifiers: >B (1 byte unsigned), >H (2 byte unsigned), >I (4 byte unsigned), >i (4 byte signed), >q (8 byte signed).
Help me debug a syntax error in my protocol.py wire protocol module
Verify with a round-trip test
A protocol is only correct if you can encode a message, decode it, and get back the original fields. This round-trip property is the fundamental guarantee your broker depends on.
- Confirm the encode/decode cycle works by running this command from inside your minikafka/ directory:
python3 -c "
from protocol import *
req = ProduceRequest(topic='orders', partition=2, key=b'user-1', value=b'buy 100 shares')
encoded = encode_produce_request(req)
decoded = decode_request(encoded)
assert decoded.topic == req.topic
assert decoded.partition == req.partition
assert decoded.key == req.key
assert decoded.value == req.value
print(f'Encoded {len(encoded)} bytes, decoded back: topic={decoded.topic} partition={decoded.partition} key={decoded.key} value={decoded.value}')
print('Round-trip test passed!')
"- You should see output showing the encoded byte count and the decoded fields matching the originals, followed by Round-trip test passed!.
Seeing an import error or assertion failure?
- Make sure you are running the command from inside the minikafka/ directory where protocol.py lives.
- If you see an AssertionError, one of the decoded fields does not match the input. Double-check that encode_produce_request and decode_request use the same field order and format specifiers.
Help me fix my round-trip test failure
✔️ Awesome, I've got everything!
Great. Double check you have saved protocol.py before moving on.
ⓧ I'd like to double check the full code
"""Binary wire protocol for MiniKafka.
Frame format: [4B length][1B api_key][payload...]
All integers are big-endian.
"""
import struct
import socket
from dataclasses import dataclass, field
API_PRODUCE = 0
API_FETCH = 1
@dataclass
class ProduceRequest:
topic: str
partition: int
key: bytes | None
value: bytes
@dataclass
class FetchRequest:
topic: str
partition: int
offset: int
max_bytes: int
@dataclass
class ProduceResponse:
offset: int
error: int = 0
@dataclass
class RecordData:
offset: int
key: bytes | None
value: bytes
@dataclass
class FetchResponse:
records: list[RecordData] = field(default_factory=list)
error: int = 0
def recv_exact(sock: socket.socket, n: int) -> bytes:
"""Read exactly n bytes from socket."""
buf = bytearray()
while len(buf) < n:
chunk = sock.recv(n - len(buf))
if not chunk:
raise ConnectionError("Connection closed")
buf.extend(chunk)
return bytes(buf)
def recv_frame(sock: socket.socket) -> bytes:
"""Read a length-prefixed frame."""
length_bytes = recv_exact(sock, 4)
length = struct.unpack(">I", length_bytes)[0]
return recv_exact(sock, length)
def send_frame(sock: socket.socket, payload: bytes) -> None:
"""Send a length-prefixed frame."""
sock.sendall(struct.pack(">I", len(payload)) + payload)
def encode_produce_request(req: ProduceRequest) -> bytes:
topic_bytes = req.topic.encode()
key_bytes = req.key if req.key is not None else b""
key_len = len(req.key) if req.key is not None else -1
buf = struct.pack(">B", API_PRODUCE)
buf += struct.pack(">H", len(topic_bytes)) + topic_bytes
buf += struct.pack(">I", req.partition)
buf += struct.pack(">i", key_len)
if key_len >= 0:
buf += key_bytes
buf += struct.pack(">I", len(req.value)) + req.value
return buf
def encode_fetch_request(req: FetchRequest) -> bytes:
topic_bytes = req.topic.encode()
buf = struct.pack(">B", API_FETCH)
buf += struct.pack(">H", len(topic_bytes)) + topic_bytes
buf += struct.pack(">I", req.partition)
buf += struct.pack(">q", req.offset)
buf += struct.pack(">I", req.max_bytes)
return buf
def encode_produce_response(resp: ProduceResponse) -> bytes:
return struct.pack(">Bq", resp.error, resp.offset)
def encode_fetch_response(resp: FetchResponse) -> bytes:
buf = struct.pack(">BI", resp.error, len(resp.records))
for rec in resp.records:
key_bytes = rec.key if rec.key is not None else b""
key_len = len(rec.key) if rec.key is not None else -1
buf += struct.pack(">q", rec.offset)
buf += struct.pack(">i", key_len)
if key_len >= 0:
buf += key_bytes
buf += struct.pack(">I", len(rec.value)) + rec.value
return buf
def decode_request(data: bytes) -> ProduceRequest | FetchRequest:
pos = 0
api_key = struct.unpack_from(">B", data, pos)[0]
pos += 1
topic_len = struct.unpack_from(">H", data, pos)[0]
pos += 2
topic = data[pos : pos + topic_len].decode()
pos += topic_len
partition = struct.unpack_from(">I", data, pos)[0]
pos += 4
if api_key == API_PRODUCE:
key_len = struct.unpack_from(">i", data, pos)[0]
pos += 4
key = None
if key_len >= 0:
key = data[pos : pos + key_len]
pos += key_len
value_len = struct.unpack_from(">I", data, pos)[0]
pos += 4
value = data[pos : pos + value_len]
return ProduceRequest(topic, partition, key, value)
elif api_key == API_FETCH:
offset = struct.unpack_from(">q", data, pos)[0]
pos += 8
max_bytes = struct.unpack_from(">I", data, pos)[0]
return FetchRequest(topic, partition, offset, max_bytes)
raise ValueError(f"Unknown API key: {api_key}")
def decode_produce_response(data: bytes) -> ProduceResponse:
error, offset = struct.unpack(">Bq", data)
return ProduceResponse(offset=offset, error=error)
def decode_fetch_response(data: bytes) -> FetchResponse:
pos = 0
error, num_records = struct.unpack_from(">BI", data, pos)
pos += 5
records = []
for _ in range(num_records):
offset = struct.unpack_from(">q", data, pos)[0]
pos += 8
key_len = struct.unpack_from(">i", data, pos)[0]
pos += 4
key = None
if key_len >= 0:
key = data[pos : pos + key_len]
pos += key_len
value_len = struct.unpack_from(">I", data, pos)[0]
pos += 4
value = data[pos : pos + value_len]
pos += value_len
records.append(RecordData(offset=offset, key=key, value=value))
return FetchResponse(records=records, error=error)Your wire protocol is complete. Next up, you will build the broker that listens on a TCP socket and uses these encode/decode functions to handle live client connections.
Build the Broker
You have a storage engine that writes records to disk and a wire protocol that encodes requests into bytes. Now you need the piece that sits between them: the broker.
The broker accepts TCP connections from clients, decodes their requests using the protocol module, routes each request to the correct topic-partition CommitLog, and sends back the response. The entire job is glue code, but it is the glue that turns two libraries into a running service.
In this step, get ready to:
- Write the Broker class with auto-creating log management and request handlers.
- Implement a TCP server loop that spawns a thread per connection.
- Start the broker and confirm it binds to port 9092.
Implement request dispatch
The broker needs to maintain a registry of commit logs, one per topic-partition pair. When a produce or fetch request arrives, it looks up (or auto-creates) the appropriate log and delegates the operation.
- Open broker.py in Cursor and add the imports, constants, and the first half of the Broker class by pasting this code:
"""MiniKafka broker: TCP server dispatching produce/fetch to commit logs."""
import socket
import threading
from pathlib import Path
from storage import CommitLog, Record
from protocol import (
recv_frame,
send_frame,
decode_request,
encode_produce_response,
encode_fetch_response,
ProduceRequest,
FetchRequest,
ProduceResponse,
FetchResponse,
RecordData,
)
# Broker configuration
DATA_DIR = Path("data")
HOST = "127.0.0.1"
PORT = 9092
class Broker:
def __init__(self, data_dir: Path = DATA_DIR):
self.data_dir = data_dir
self.data_dir.mkdir(parents=True, exist_ok=True)
# Registry of commit logs keyed by (topic, partition)
self._logs: dict[tuple[str, int], CommitLog] = {}
self._logs_lock = threading.Lock()
def get_log(self, topic: str, partition: int) -> CommitLog:
"""Look up or auto-create a CommitLog for the given topic-partition."""
key = (topic, partition)
with self._logs_lock:
if key not in self._logs:
self._logs[key] = CommitLog(self.data_dir, topic, partition)
return self._logs[key]
def handle_produce(self, req: ProduceRequest) -> ProduceResponse:
"""Append the record to the correct log and return the assigned offset."""
log = self.get_log(req.topic, req.partition)
offset = log.append(req.key, req.value)
return ProduceResponse(offset=offset)
def handle_fetch(self, req: FetchRequest) -> FetchResponse:
"""Read records from the requested offset and return them."""
log = self.get_log(req.topic, req.partition)
records = log.read(req.offset, req.max_bytes)
record_data = [RecordData(r.offset, r.key, r.value) for r in records]
return FetchResponse(records=record_data)What does this code do?
The broker maintains a dictionary _logs that maps each (topic, partition) tuple to a CommitLog instance. The get_log method auto-creates a new log on first access, so topics spring into existence the moment a client produces to them.
handle_produce appends the record and returns the assigned offset. handle_fetch reads records starting at the consumer's requested offset and converts them into wire-friendly RecordData objects for the response encoder.
You won't see a change yet. The next chunk adds the TCP server that actually starts the broker.
Why is the broker stateless about consumer offsets?
Notice that the broker never tracks where a consumer is in the log. Offsets are entirely consumer-controlled. This is Apache Kafka's fundamental architectural insight: because the broker just stores and forwards bytes, it stays simple and scalable.
Consumers can replay messages by seeking backwards. They can skip ahead by advancing their offset. The broker does not need to know or care. This also means topics auto-create on first produce, just like Kafka's auto.create.topics.enable setting.
Implement the TCP server loop
The broker needs to accept incoming TCP connections and handle each one concurrently. Each connection thread loops forever, reading frames, dispatching requests, and sending responses until the client disconnects.
- Below the handle_fetch method, still inside the Broker class, add the connection handler and server startup by pasting this code:
def handle_connection(self, conn: socket.socket, addr: tuple) -> None:
"""Process requests from a single client connection in a loop."""
print(f"[broker] connection from {addr}")
try:
while True:
# Read one frame, decode, dispatch, respond
frame = recv_frame(conn)
request = decode_request(frame)
if isinstance(request, ProduceRequest):
resp = self.handle_produce(request)
send_frame(conn, encode_produce_response(resp))
elif isinstance(request, FetchRequest):
resp = self.handle_fetch(request)
send_frame(conn, encode_fetch_response(resp))
except ConnectionError:
pass
finally:
conn.close()
print(f"[broker] {addr} disconnected")
def start(self) -> None:
"""Bind to the configured port and accept connections forever."""
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.bind((HOST, PORT))
srv.listen()
print(f"[broker] listening on {HOST}:{PORT}")
try:
while True:
conn, addr = srv.accept()
# Daemon thread dies when main thread exits
t = threading.Thread(target=self.handle_connection, args=(conn, addr), daemon=True)
t.start()
except KeyboardInterrupt:
print("\n[broker] shutting down")
finally:
srv.close()
if __name__ == "__main__":
Broker().start()What does this code do?
handle_connection runs in its own thread for each connected client. It loops on recv_frame (from your protocol module), decodes the request, calls the appropriate handler, and sends back the encoded response. When the client disconnects, the ConnectionError exception breaks the loop cleanly.
start binds a TCP socket to port 9092 (Kafka's default port) and spawns a daemon thread for each accepted connection. SO_REUSEADDR prevents "address already in use" errors when you restart the broker quickly.
- Save broker.py.
- Start the broker by running this command from your minikafka/ directory:
python3 broker.py- You should see [broker] listening on 127.0.0.1:9092 printed to your terminal. The process stays running, waiting for client connections.
- Press Ctrl+C to stop the broker for now. You will start it again in the next step when you have clients ready.
Seeing 'Address already in use'?
This means a previous broker process is still running on port 9092. Find and kill it, or wait a few seconds for the OS to release the port. On macOS/Linux, run lsof -i :9092 to identify the process.
If you see an ImportError, make sure you are running the command from inside the minikafka/ directory where storage.py and protocol.py live.
Help me debug my broker startup error
✔️ Awesome, I've got everything!
Great. Double check you have saved broker.py and that it starts cleanly before moving on.
ⓧ I'd like to double check the full code
Here is the complete broker.py file for comparison:
"""MiniKafka broker: TCP server dispatching produce/fetch to commit logs."""
import socket
import threading
from pathlib import Path
from storage import CommitLog, Record
from protocol import (
recv_frame,
send_frame,
decode_request,
encode_produce_response,
encode_fetch_response,
ProduceRequest,
FetchRequest,
ProduceResponse,
FetchResponse,
RecordData,
)
DATA_DIR = Path("data")
HOST = "127.0.0.1"
PORT = 9092
class Broker:
def __init__(self, data_dir: Path = DATA_DIR):
self.data_dir = data_dir
self.data_dir.mkdir(parents=True, exist_ok=True)
self._logs: dict[tuple[str, int], CommitLog] = {}
self._logs_lock = threading.Lock()
def get_log(self, topic: str, partition: int) -> CommitLog:
key = (topic, partition)
with self._logs_lock:
if key not in self._logs:
self._logs[key] = CommitLog(self.data_dir, topic, partition)
return self._logs[key]
def handle_produce(self, req: ProduceRequest) -> ProduceResponse:
log = self.get_log(req.topic, req.partition)
offset = log.append(req.key, req.value)
return ProduceResponse(offset=offset)
def handle_fetch(self, req: FetchRequest) -> FetchResponse:
log = self.get_log(req.topic, req.partition)
records = log.read(req.offset, req.max_bytes)
record_data = [RecordData(r.offset, r.key, r.value) for r in records]
return FetchResponse(records=record_data)
def handle_connection(self, conn: socket.socket, addr: tuple) -> None:
print(f"[broker] connection from {addr}")
try:
while True:
frame = recv_frame(conn)
request = decode_request(frame)
if isinstance(request, ProduceRequest):
resp = self.handle_produce(request)
send_frame(conn, encode_produce_response(resp))
elif isinstance(request, FetchRequest):
resp = self.handle_fetch(request)
send_frame(conn, encode_fetch_response(resp))
except ConnectionError:
pass
finally:
conn.close()
print(f"[broker] {addr} disconnected")
def start(self) -> None:
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.bind((HOST, PORT))
srv.listen()
print(f"[broker] listening on {HOST}:{PORT}")
try:
while True:
conn, addr = srv.accept()
t = threading.Thread(target=self.handle_connection, args=(conn, addr), daemon=True)
t.start()
except KeyboardInterrupt:
print("\n[broker] shutting down")
finally:
srv.close()
if __name__ == "__main__":
Broker().start()Your broker is alive and ready to accept connections. In the next step, you will build Producer and Consumer clients that speak the wire protocol and stream messages through this broker end-to-end.
Build the Clients and Demo
Your broker is listening on port 9092, ready to accept connections. But without clients that speak the binary protocol, it has nothing to talk to.
In this step, you'll build a Producer that deterministically routes messages to partitions using key hashing, and a Consumer that tracks its own offset for replay. Then you'll wire them together in a demo that proves end-to-end streaming works.
In this step, get ready to:
- Implement a Producer with key-based partition routing and round-robin fallback.
- Implement a Consumer with offset tracking and seek-based replay.
- Write a demo that produces, consumes, and replays messages across partitions.
Implement the Producer with key-based partition routing
In Kafka, the producer decides which partition a message lands in. Messages with the same key always go to the same partition, which guarantees ordering for that key. Messages without a key get distributed round-robin.
- Open client.py in Cursor and add the imports, constants, and Producer class by pasting this code:
"""MiniKafka producer and consumer clients."""
import socket
import zlib
from dataclasses import dataclass
from protocol import (
ProduceRequest,
FetchRequest,
RecordData,
encode_produce_request,
encode_fetch_request,
decode_produce_response,
decode_fetch_response,
send_frame,
recv_frame,
)
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 9092
class Producer:
def __init__(self, host: str = DEFAULT_HOST, port: int = DEFAULT_PORT, num_partitions: int = 4):
self.host = host
self.port = port
self.num_partitions = num_partitions
self._round_robin = 0
self._sock: socket.socket | None = None
def _connect(self) -> socket.socket:
# Reuse a single TCP connection for all sends
if self._sock is None:
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self._sock.connect((self.host, self.port))
return self._sock
def _partition_for(self, key: bytes | None) -> int:
# Deterministic hash routing for keyed messages
if key is not None:
return zlib.crc32(key) % self.num_partitions
# Round-robin for null keys
p = self._round_robin % self.num_partitions
self._round_robin += 1
return p
def send(self, topic: str, value: bytes, key: bytes | None = None) -> int:
# Route to partition, encode request, send, return assigned offset
partition = self._partition_for(key)
req = ProduceRequest(topic=topic, partition=partition, key=key, value=value)
sock = self._connect()
send_frame(sock, encode_produce_request(req))
resp_data = recv_frame(sock)
resp = decode_produce_response(resp_data)
return resp.offset
def close(self) -> None:
if self._sock:
self._sock.close()
self._sock = NoneWhat does this code do?
The Producer class handles three responsibilities:
- _partition_for uses zlib.crc32(key) % num_partitions to deterministically route keyed messages. The same key always hashes to the same partition, guaranteeing message ordering for that key.
- _connect lazily opens one persistent TCP connection to the broker. This avoids the overhead of reconnecting on every send.
- send encodes a ProduceRequest using your wire protocol, sends it over the socket, and returns the offset the broker assigned.
Why CRC32 instead of murmur2?
Real Kafka uses a murmur2 hash for partition routing. Our implementation uses zlib.crc32 because it is available in Python's standard library and is deterministic across runs. The concept is identical: a hash function that maps a key to a fixed partition number.
- Save client.py.
Seeing an import error?
- Make sure protocol.py is in the same directory as client.py. The import expects both files in the minikafka/ folder.
- Confirm you're using Python 3.10+ because the union type syntax socket.socket | None requires it.
Help me debug my Producer import errors in client.py
Implement the Consumer with offset tracking
The consumer controls its own position in the log. This is Kafka's fundamental insight that makes brokers simple and scalable: the broker just stores bytes, and consumers decide where to read from.
- Add the Consumer class below the Producer class in client.py by pasting this code at the bottom of the file:
class Consumer:
def __init__(self, host: str = DEFAULT_HOST, port: int = DEFAULT_PORT):
self.host = host
self.port = port
self._offsets: dict[tuple[str, int], int] = {}
self._sock: socket.socket | None = None
def _connect(self) -> socket.socket:
# Reuse a persistent TCP connection
if self._sock is None:
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self._sock.connect((self.host, self.port))
return self._sock
def seek(self, topic: str, partition: int, offset: int) -> None:
# Reset position to enable replay from any offset
self._offsets[(topic, partition)] = offset
def poll(self, topic: str, partition: int, max_bytes: int = 65536) -> list[RecordData]:
# Fetch from current offset and advance position
offset = self._offsets.get((topic, partition), 0)
req = FetchRequest(topic=topic, partition=partition, offset=offset, max_bytes=max_bytes)
sock = self._connect()
send_frame(sock, encode_fetch_request(req))
resp_data = recv_frame(sock)
resp = decode_fetch_response(resp_data)
if resp.records:
self._offsets[(topic, partition)] = resp.records[-1].offset + 1
return resp.records
def close(self) -> None:
if self._sock:
self._sock.close()
self._sock = NoneWhat does this code do?
- _offsets is a dictionary keyed by (topic, partition) that tracks the consumer's position in each partition. The broker never stores this state.
- poll fetches records starting from the current offset, then advances the offset past the last record received. This pull-based model lets the consumer control its own pace.
- seek resets the offset to any position. This enables replay, which is one of Kafka's most powerful features. A consumer can re-read old messages at any time.
- Save client.py.
Why does the consumer track its own offset?
Traditional message queues track which messages have been delivered to each consumer. This creates broker-side state that grows with the number of consumers. Kafka's pull-based design moves that responsibility to the consumer, letting the broker stay stateless and scale horizontally.
Consumer not receiving records?
- Make sure you call seek() before poll() to set the starting offset. Without it, the offset defaults to 0, which may be correct, but verify the topic-partition has data.
- Make sure the broker is running before the consumer connects. A ConnectionRefusedError means the broker is not listening on port 9092.
Help me debug why my Consumer isn't receiving records
✔️ Awesome, I've got everything!
Great. Double check you've saved client.py with both the Producer and Consumer classes.
ⓧ I'd like to double check the full code
"""MiniKafka producer and consumer clients."""
import socket
import zlib
from dataclasses import dataclass
from protocol import (
ProduceRequest,
FetchRequest,
RecordData,
encode_produce_request,
encode_fetch_request,
decode_produce_response,
decode_fetch_response,
send_frame,
recv_frame,
)
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 9092
class Producer:
def __init__(self, host: str = DEFAULT_HOST, port: int = DEFAULT_PORT, num_partitions: int = 4):
self.host = host
self.port = port
self.num_partitions = num_partitions
self._round_robin = 0
self._sock: socket.socket | None = None
def _connect(self) -> socket.socket:
if self._sock is None:
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self._sock.connect((self.host, self.port))
return self._sock
def _partition_for(self, key: bytes | None) -> int:
if key is not None:
return zlib.crc32(key) % self.num_partitions
p = self._round_robin % self.num_partitions
self._round_robin += 1
return p
def send(self, topic: str, value: bytes, key: bytes | None = None) -> int:
partition = self._partition_for(key)
req = ProduceRequest(topic=topic, partition=partition, key=key, value=value)
sock = self._connect()
send_frame(sock, encode_produce_request(req))
resp_data = recv_frame(sock)
resp = decode_produce_response(resp_data)
return resp.offset
def close(self) -> None:
if self._sock:
self._sock.close()
self._sock = None
class Consumer:
def __init__(self, host: str = DEFAULT_HOST, port: int = DEFAULT_PORT):
self.host = host
self.port = port
self._offsets: dict[tuple[str, int], int] = {}
self._sock: socket.socket | None = None
def _connect(self) -> socket.socket:
if self._sock is None:
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self._sock.connect((self.host, self.port))
return self._sock
def seek(self, topic: str, partition: int, offset: int) -> None:
self._offsets[(topic, partition)] = offset
def poll(self, topic: str, partition: int, max_bytes: int = 65536) -> list[RecordData]:
offset = self._offsets.get((topic, partition), 0)
req = FetchRequest(topic=topic, partition=partition, offset=offset, max_bytes=max_bytes)
sock = self._connect()
send_frame(sock, encode_fetch_request(req))
resp_data = recv_frame(sock)
resp = decode_fetch_response(resp_data)
if resp.records:
self._offsets[(topic, partition)] = resp.records[-1].offset + 1
return resp.records
def close(self) -> None:
if self._sock:
self._sock.close()
self._sock = NoneWrite the end-to-end demo
The demo ties everything together: producing keyed messages across partitions, consuming them in order, and replaying from a specific offset.
- Open demo.py and paste the following code:
"""End-to-end demonstration of MiniKafka."""
import time
from client import Producer, Consumer
TOPIC = "user-events"
NUM_PARTITIONS = 4
def main():
producer = Producer(num_partitions=NUM_PARTITIONS)
# Produce 20 messages with user-ID keys for deterministic routing
print("=== Producing 20 messages with user-ID keys ===\n")
for i in range(20):
user_id = f"user-{i % 5}"
value = f"event-{i} from {user_id}"
offset = producer.send(TOPIC, value.encode(), key=user_id.encode())
partition = producer._partition_for(user_id.encode())
print(f" [{TOPIC}-{partition}] offset={offset} key={user_id} value={value}")
producer.close()
# Consume all partitions from the beginning
print("\n=== Consuming all partitions from offset 0 ===\n")
consumer = Consumer()
for p in range(NUM_PARTITIONS):
consumer.seek(TOPIC, p, 0)
records = consumer.poll(TOPIC, p)
if records:
print(f" Partition {p} ({len(records)} records):")
for r in records:
key_str = r.key.decode() if r.key else "null"
print(f" offset={r.offset} key={key_str} value={r.value.decode()}")
else:
print(f" Partition {p}: empty")
consumer.close()
# Demonstrate replay by seeking back to offset 0
print("\n=== Replay: seeking partition 0 back to offset 0 ===\n")
consumer = Consumer()
consumer.seek(TOPIC, 0, 0)
records = consumer.poll(TOPIC, 0)
print(f" Re-read {len(records)} records from partition 0")
consumer.close()
if __name__ == "__main__":
main()What does this demo prove?
- Key-based routing: all messages from user-0 always land in the same partition. You'll see repeated keys grouped together in the consumption output.
- Per-partition ordering: within each partition, messages appear in the exact order they were produced. This is Kafka's ordering guarantee.
- Replay: the consumer seeks back to offset 0 and re-reads partition 0, proving that messages are durably stored and replayable at any time.
- Save demo.py.
Now run the full system. You need two terminal windows for this.
- In the first terminal, make sure you are in the minikafka/ directory and start the broker by running:
python broker.py- You should see [broker] listening on 127.0.0.1:9092 printed in the terminal.
- Open a second terminal window, navigate to the same minikafka/ directory, and run the demo by running:
python demo.py- You should see 20 messages produced with their partition assignments and offsets, followed by a consumption section showing records grouped by partition, and finally a replay of partition 0.
Connection refused when running demo.py?
- Make sure the broker is running in the other terminal. You need python broker.py active before you run the demo.
- If you previously ran the broker and killed it, port 9092 might still be in use. Wait a few seconds or check for lingering processes.
- If you see a ModuleNotFoundError, make sure both terminals are in the minikafka/ directory where all your .py files live.
Help me debug my MiniKafka demo connection issue
✔️ Awesome, I've got everything!
Great. You can stop the broker with Ctrl+C when you're done inspecting the output.
ⓧ I'd like to double check the full code
"""End-to-end demonstration of MiniKafka."""
import time
from client import Producer, Consumer
TOPIC = "user-events"
NUM_PARTITIONS = 4
def main():
producer = Producer(num_partitions=NUM_PARTITIONS)
print("=== Producing 20 messages with user-ID keys ===\n")
for i in range(20):
user_id = f"user-{i % 5}"
value = f"event-{i} from {user_id}"
offset = producer.send(TOPIC, value.encode(), key=user_id.encode())
partition = producer._partition_for(user_id.encode())
print(f" [{TOPIC}-{partition}] offset={offset} key={user_id} value={value}")
producer.close()
print("\n=== Consuming all partitions from offset 0 ===\n")
consumer = Consumer()
for p in range(NUM_PARTITIONS):
consumer.seek(TOPIC, p, 0)
records = consumer.poll(TOPIC, p)
if records:
print(f" Partition {p} ({len(records)} records):")
for r in records:
key_str = r.key.decode() if r.key else "null"
print(f" offset={r.offset} key={key_str} value={r.value.decode()}")
else:
print(f" Partition {p}: empty")
consumer.close()
print("\n=== Replay: seeking partition 0 back to offset 0 ===\n")
consumer = Consumer()
consumer.seek(TOPIC, 0, 0)
records = consumer.poll(TOPIC, 0)
print(f" Re-read {len(records)} records from partition 0")
consumer.close()
if __name__ == "__main__":
main()You now have a complete message streaming system. A producer routes messages by key, a broker stores them in segmented commit logs, and a consumer replays them at will. Next up: extend the system with consumer group coordination in the Secret Mission.
Secret mission
Add Consumer Group Coordination
Your broker currently treats every consumer as independent. In real Kafka, consumers form groups where each partition is consumed by exactly one member. Can you add broker-side group coordination with range-based partition assignment and automatic rebalancing?
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally with no cloud resources or paid services. There are no ongoing costs.
Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- The minikafka/ project directory on your Desktop (contains all Python source files).
- The minikafka/data/ directory (commit log segment files created at runtime by the broker).
Keep everything running
No action needed. Choose this if you plan to keep experimenting with your broker, adding new API types, or extending the consumer group protocol.
- Your source files stay in ~/Desktop/minikafka/.
- The data/ directory holds your commit log segments from previous broker runs.
- If the broker is still running in a terminal, it will continue to listen on port 9092 until you press Ctrl+C.
Pause - I'll come back to this later
Shut down the broker process to free up the port. Your files and log data stay on disk for next time.
- Switch to the terminal where the broker is running and press Ctrl+C to stop it.
- Your source files and data/ directory remain intact.
- When you come back, start the broker again with python broker.py from inside the minikafka/ directory. It will recover existing log state on startup.
Delete - I don't want to use this again
Remove all project files and start fresh. This deletes your source code and all commit log data.
- If the broker is still running, press Ctrl+C in its terminal to stop it.
- Delete the entire project directory by running this command:
macOS
rm -rf ~/Desktop/minikafkaWindows
rmdir /s /q %USERPROFILE%\Desktop\minikafka- Confirm the folder is gone by running:
ls ~/DesktopYou should no longer see minikafka in the output.
That's a Wrap!
That's a Wrap!
Nice work. You just built a message streaming engine from scratch, implementing the same fundamental architecture that powers Apache Kafka at production scale.
You've learned how to:
- Build an append-only commit log storage engine with segment rotation and CRC32 integrity validation.
- Design a binary wire protocol with length-prefixed framing and wire it to a TCP broker handling concurrent client connections.
- Implement key-based partition routing in producers and offset-based replay in consumers for deterministic message ordering.
- Secret Mission: Added consumer group coordination with range-based partition assignment and rebalancing.
Ready to quiz yourself?