But Why??
I had used Kafka enough to know the terminology: topics, partitions, producers, consumers, replication. But knowing the pieces is different from understanding why they fit together the way they do.
So I decided to build a lightweight Kafka-like message broker from scratch. Not as a replacement for Kafka, but as a way to understand what actually happens underneath the abstractions.
Why does a partition need a leader? Why can't every replica accept writes? What actually happens on disk when a message is produced? Why does adding a consumer to a group trigger a rebalance?
I knew the answers in theory, but I couldn't explain them the way you can explain something you've actually built.
So I built kafka-lite, a small Kafka-inspired message broker from scratch.
The broker is written in Go. It uses TCP for communication, file I/O for persistence, and a small Python client SDK. I deliberately avoided using an existing embedded database for the log because I wanted to understand what the storage layer was actually doing.
This article goes through what I built, stage by stage, along with some of the design decisions, bugs, and simplifications I made along the way.
This is not a production Kafka implementation. It is called kafka-lite for a reason. Where I simplified something, I'll explain what I changed and what that means.
What I was trying to build
Before writing code, I reduced Kafka to a few ideas I wanted to understand.
At the center is a durable, ordered log. Producers append messages to the log, and consumers read them later. Consumers don't have to process a message immediately after it is produced.
Then comes partitioning. A topic can be divided into multiple partitions, allowing different parts of the topic to be processed independently and in parallel.
Finally, there is distribution. Partitions can live on different brokers, and their data can be replicated so that losing one broker does not necessarily mean losing the data.
The other Kafka concepts I wanted to understand, such as consumer groups, offsets, rebalancing, replication, and leader election, build around these ideas.
I broke the project into five stages:
Stage 1: Single broker, TCP protocol, persistent log, topics and partitions
Stage 2: Log segments and retention
Stage 3: Consumer groups, offsets and rebalancing
Stage 4: Multiple brokers, leadership, replication and failover
Stage 5: Python SDK, Docker Compose and integration tests
Each stage was built on top of the previous one.
I also tried to test the system by actually running it. That meant opening real TCP connections, starting multiple broker processes, killing processes, and checking what happened afterward.
That turned out to be where a lot of the interesting bugs appeared.
Stage 1: A broker is a TCP server with opinions about files
The wire protocol
Before a producer can send a message, the client and broker need to agree on how the bytes travelling over the TCP connection should be interpreted.
Kafka has a custom binary protocol. I decided not to start there.
For kafka-lite, every request is simply:
[4-byte length][JSON payload]
The length tells the receiver how many bytes belong to the message, and the JSON payload contains the actual request.
The Go implementation looks like this:
func WriteFrame(w io.Writer, msgType MsgType, v interface{}) error {
payload, err := json.Marshal(v)
if err != nil {
return fmt.Errorf("protocol: marshal payload: %w", err)
}
env := Envelope{Type: msgType, Payload: payload}
data, err := json.Marshal(env)
if err != nil {
return fmt.Errorf("protocol: marshal envelope: %w", err)
}
lenBuf := make([]byte, 4)
binary.BigEndian.PutUint32(lenBuf, uint32(len(data)))
if _, err := w.Write(lenBuf); err != nil {
return fmt.Errorf("protocol: write length prefix: %w", err)
}
if _, err := w.Write(data); err != nil {
return fmt.Errorf("protocol: write payload: %w", err)
}
return nil
}
The reason for choosing JSON was that it made development and debugging easier. I could connect to the broker with a simple tool, inspect the request, and understand what was being sent without having to decode a binary protocol.
That is not a good choice if the goal is to maximize throughput, but that wasn't the goal here. I wanted the protocol to be easy to understand while I was building the rest of the system.
I also kept the framing code isolated, so replacing JSON with a binary encoding later would not require changing the whole broker.
The log: where the database part lives
This was one of the parts I was most interested in.
It is easy to think of Kafka as a service that writes messages to disk. Actually implementing that storage layer makes the idea much more clear.
In kafka-lite, a partition is represented by a directory containing log files.
A record on disk looks like this:
[8 bytes offset]
[8 bytes timestamp]
[4 bytes key length]
[key bytes]
[4 bytes value length]
[value bytes]
Appending a record is relatively straightforward:
func (l *Log) Append(key, value []byte) (uint64, error) {
l.mu.Lock()
defer l.mu.Unlock()
active := l.active()
if active.size > 0 && active.size >= l.opts.maxSegmentBytes() {
if err := l.roll(); err != nil {
return 0, err
}
active = l.active()
}
offset := active.nextOffset()
ts := time.Now().UnixNano()
header := make([]byte, recordHeaderSize)
binary.BigEndian.PutUint64(header[0:8], offset)
binary.BigEndian.PutUint64(header[8:16], uint64(ts))
binary.BigEndian.PutUint32(header[16:20], uint32(len(key)))
buf := make([]byte, 0, recordHeaderSize+len(key)+lenFieldSize+len(value))
buf = append(buf, header...)
buf = append(buf, key...)
valLenBuf := make([]byte, lenFieldSize)
binary.BigEndian.PutUint32(valLenBuf, uint32(len(value)))
buf = append(buf, valLenBuf...)
buf = append(buf, value...)
pos, err := active.file.Seek(0, io.SeekEnd)
if err != nil {
return 0, err
}
if _, err := active.file.Write(buf); err != nil {
return 0, err
}
if err := active.file.Sync(); err != nil {
return 0, err
}
active.index = append(active.index, indexEntry{position: pos})
active.size = pos + int64(len(buf))
active.newestTS = ts
return offset, nil
}
The important part here is:
active.file.Sync()
That asks the operating system to flush the data to stable storage before the append is considered complete.
I benchmarked this instead of just assuming it was expensive. On my setup, appending one record took roughly 614 microseconds, while reading one took around 12 microseconds.
The difference was significant. Reads are relatively cheap because they are basically a seek and read, and the operating system may already have the data cached. But a synchronous disk flush explicitly asks the system to make the data durable before continuing.
That helped me understand why systems like Kafka have different durability and batching options. There is a real tradeoff between throughput and how much work you are willing to lose during a failure.
Crash recovery
This raised another question. What happens if the process dies while it is writing a record?
The result can be a file containing several complete records followed by an incomplete record. For example:
record
record
record
half-written record
On startup, kafka-lite scans the log and checks that records are complete and that offsets are sequential. If it encounters an invalid record, it truncates the file at that position.
The basic idea is:
if offset != expectedOffset {
break
}
// ...
l.file.Truncate(pos)
The important part is that the broker does not try to interpret the incomplete record as valid data.
I wrote a test that deliberately corrupted a log in the middle of a record and then restarted the broker. The broker recovered the valid portion of the log and removed the incomplete data.
That was a useful distinction for the system.
Partitioning
A topic in kafka-lite can have multiple partitions. Each message needs to be assigned to one of them.
If a key is provided, it hashes the key and uses the result to choose the partition:
func (t *Topic) choosePartition(key []byte) int {
if key != nil {
h := fnv.New32a()
h.Write(key)
return int(h.Sum32()) % t.NumPartitions
}
idx := atomic.AddUint64(&t.roundRobin, 1)
return int(idx-1) % t.NumPartitions
}
If there is no key, messages are distributed using round-robin assignment.
The key-based approach gives an important property:
same key -> same partition
So if several messages belong to the same user, account, order, or other entity, they can be kept in the same partition and therefore retain their relative ordering.
That also made an important Kafka concept much clearer to me: ordering is a partition-level property, not a topic-level property. A topic with four partitions does not have one global order across all four partitions.
By the end of this, I had a broker that could create topics, accept messages over TCP, persist them to disk, and read them back after a restart.
Stage 2: The log can't grow forever
At this point there was an obvious problem. Each partition had one log file, and that file would keep growing forever. Disks are not infinite, and old messages eventually stop being useful.
The solution was log segments. Instead of having one enormous file, each partition contains a sequence of smaller files. For example:
00000000000000000000.log
00000000000000000047.log
00000000000000000094.log
...
The number represents the offset of the first record in the segment. Only the newest segment is actively being written to.
When the active segment reaches its size limit, the broker creates a new one:
func (l *Log) Append(key, value []byte) (uint64, error) {
active := l.active()
if active.size >= l.opts.maxSegmentBytes() {
l.roll()
active = l.active()
}
// Append record...
}
This makes retention much simpler. Instead of deleting individual records from a large file, the broker can remove entire old segments. For example:
segment 1 <- old
segment 2 <- old
segment 3
segment 4 <- active
If segment 1 and segment 2 are past the retention limit, they can simply be deleted.
The retention logic looks roughly like this:
I'm leaving out the full implementation here to keep the article from getting too long But the complete code is available in the GitHub repo linked at the beginning.
func (l *Log) ApplyRetention(now time.Time) (int, error) {
for len(l.segments) > 1 {
oldest := l.segments[0]
if now.Sub(time.Unix(0, oldest.newestTS)) < l.opts.RetentionAge {
break
}
l.deleteOldestSegment()
}
return 0, nil
}
I deliberately used very small segment sizes while testing this. With 2 KB segments and a 5 KB retention limit, I produced 100 small records and watched the log create multiple segments.
After retention ran, the number of segments dropped and the partition's starting offset moved forward. Trying to read an expired offset returned an error instead of silently returning incorrect data.
Stage 3: Consumer groups
This was the stage where I started to build something resembling a distributed system rather than just a networked storage service.
A consumer group is a collection of consumers sharing the work of reading a topic. For example, suppose a topic has four partitions:
partition 0
partition 1
partition 2
partition 3
With one consumer:
consumer A -> 0, 1, 2, 3
With two consumers:
consumer A -> 0, 2
consumer B -> 1, 3
This allows consumers to process different partitions in parallel. But it creates a coordination problem: what happens when a consumer joins or leaves? The partition assignment has to change. That's where rebalancing comes in.
My simplified rebalance protocol
Kafka's actual consumer-group protocol is more involved. The group has a leader, and the protocol uses JoinGroup and SyncGroup phases. The leader can calculate the assignment using a configurable strategy.
I didn't implement that full protocol. In kafka-lite, the broker calculates the assignment directly:
func (c *Coordinator) rebalanceLocked(g *groupState) error {
for _, topic := range topics {
members := topicMembers[topic]
sort.Strings(members)
count, _ := c.counter.PartitionCount(topic)
for p := 0; p < count; p++ {
assignee := members[p%len(members)]
newAssignment[assignee] =
append(
newAssignment[assignee],
TopicPartition{topic, p},
)
}
}
g.generation++
return nil
}
This is much simpler. It also means kafka-lite does not support the same flexibility as real Kafka's consumer assignment strategies.
Despite the limitation, the important part is: when membership changes, partitions have to move safely between consumers.
Generation numbers
One of the more interesting problems was dealing with consumers that have stale information. For example:
Consumer A owns partition 3.
Consumer B joins.
A rebalance happens.
Partition 3 is now assigned to B.
But consumer A might not know about the rebalance immediately. It could continue processing partition 3. That can lead to two consumers processing the same partition or, worse, committing conflicting offsets.
kafka-lite uses a generation number to detect this. Every rebalance increments the generation:
func (c *Coordinator) Heartbeat(
groupID string,
memberID string,
generation int,
) (int, bool, error) {
if generation != g.generation {
return g.generation, true, ErrStaleGeneration
}
return g.generation, false, nil
}
A consumer includes the generation it believes is current in its heartbeat. If the broker has moved to a newer generation, the consumer is told to rejoin.
This made the concept of generation fencing clear to me in the context of Kafka.
Watching a rebalance happen
I also tested this manually. With four partitions:
member 1 joins "orders"
generation=1
member 1 -> orders/0
orders/1
orders/2
orders/3
Then another consumer joined:
member 2 joins "orders"
generation=2
member 1 -> orders/0
orders/2
member 2 -> orders/1
orders/3
If a consumer stopped sending heartbeats, the coordinator eventually removed it after the session timeout and triggered another rebalance.
Seeing the partitions actually move made the reason for heartbeats and session timeouts much easier to understand.
Stage 4: Making it distributed
Until this point, everything could run inside one broker. I needed multiple brokers.
This introduced two questions:
- Who is allowed to write to a partition?
- What happens when that broker dies?
Partition leadership
In a distributed system, multiple brokers may contain replicas of the same partition. But if all replicas independently accept writes, they can diverge. So there needs to be a single authority for writes to a partition. That is the leader.
For kafka-lite, I wanted to avoid introducing ZooKeeper or a consensus algorithm just for the sake of the project. So I made a major simplification.
Instead of storing a leader assignment that brokers have to agree on, they calculate the assignment deterministically. Given:
- the topic
- the partition
- the replication factor
- and the list of brokers
each broker can calculate the same replica list.
The basic idea is:
func AssignReplicas(
topic string,
partition int,
replicationFactor int,
allBrokerIDs []int,
) []int {
h := fnv.New32a()
h.Write([]byte(topic))
start := int(h.Sum32()) % len(allBrokerIDs)
replicas := make([]int, replicationFactor)
for i := range replicas {
replicas[i] =
allBrokerIDs[(start+i)%len(allBrokerIDs)]
}
return replicas
}
The leader is then the first currently alive replica:
func PickLeader(
replicas []int,
isAlive func(int) bool,
) (int, bool) {
for _, id := range replicas {
if isAlive(id) {
return id, true
}
}
return 0, false
}
This avoids a controller election and consensus protocol. But it comes with an important limitation: each broker determines liveness independently. That means two brokers can temporarily disagree about whether another broker is alive. If broker A thinks broker C is alive while broker B thinks C is dead, they could temporarily calculate different leaders.
That is a real consistency problem. Production distributed systems use coordination and consensus mechanisms precisely to handle problems like this. kafka-lite does not.
For this project, I accepted that limitation because it allowed me to explore replication and failover without first implementing an entire consensus system.
A bug caused by a default assumption
This design also gave me a real bug.
Initially, I marked every peer broker as dead until the first health check confirmed that it was alive. That sounds safe. But it wasn't.
When a cluster started, there was a short period before the first health check ran. During that period, every broker thought its peers were dead. So every broker could conclude that it was the leader for partitions it replicated.
The cluster was effectively making leadership decisions based on initialization order.
I found this through an integration test that checked whether leadership was actually distributed across brokers.
The fix was to initially assume peers were alive and only mark them dead after a health check failed:
// Assume peers are alive until a health check proves otherwise.
c.alive[p.ID] = true
In distributed systems, the default value for an unknown state can change the behavior of the entire system.
Replication
Once there are multiple brokers, data can be copied between them. When the leader accepts a message, kafka-lite sends the record to the other replicas:
func (b *Broker) replicateToFollowers(
topicName string,
topic *Topic,
partition int,
replicas []int,
rec kafkalog.Record,
) {
var wg sync.WaitGroup
for _, id := range replicas {
if id == b.brokerID || !b.cluster.IsAlive(id) {
continue
}
wg.Add(1)
go func(id int, addr string) {
defer wg.Done()
// Send record to follower.
}(id, addr)
}
wg.Wait()
}
The replicas store the same record so that another broker can continue serving the partition if the leader disappears.
There is a limitation here: kafka-lite acknowledges a write once it is durable on the leader. It does not wait for a quorum of replicas before acknowledging it. So this can happen:
producer
|
v
leader writes message
|
v
leader acknowledges producer
|
X
leader dies before replication
In that case, the acknowledged message can be lost. That is a significant difference from a quorum-based acknowledgment model.
Implementing full quorum acknowledgment would require dealing with replica confirmation, partial replication, failure cases, and recovery. I decided that was beyond the scope of this project.
Instead, I tested the case where replication had already completed. I started three brokers with replication factor 3, produced a message, waited for replication, and then killed the leader.
The surviving broker became the leader and the message was still available:
=== KILL broker 1 ===
=== fetch partition 0 from broker2 post-failover ===
offset=0 key="" value="msg-B"
=== produce new message to partition 0 post-failover ===
produced to partition=0 offset=1
That was the behavior I wanted to verify.
Redirecting clients to the leader
Another design choice was what to do when a client sends a request to the wrong broker.
I could have made that broker forward the request to the correct leader. Instead, kafka-lite returns the leader's address:
if leaderID != b.brokerID {
addr, _ := b.cluster.Addr(leaderID)
return 0, 0, &NotLeaderError{
LeaderID: leaderID,
LeaderAddr: addr,
}
}
The client then connects to the correct broker and retries.
This adds some responsibility to the client, but keeps the broker simpler. It also avoids an extra hidden network hop:
Client -> wrong broker -> leader
Instead, the client eventually communicates directly with:
Client -> leader
I implemented this behavior in both the Go CLI and the Python SDK.
While testing it, I also found a small bug in the retry logic. The same response structure was being reused for the retry, which allowed stale fields from the first response to survive into the second one. Using a fresh response structure fixed it.
Stage 5: A Python client
At this point, the broker worked, but interacting with it directly wasn't particularly convenient. So I wrote a small Python SDK.
The interesting part was that I did not share code between the Go broker and the Python client. They simply implement the same protocol. That means the two implementations have to agree on things like message types, request formats, responses, and errors.
A producer can be used like this:
class Producer:
def __init__(self, bootstrap_servers, timeout=5.0):
self.client = KafkaLiteClient(
bootstrap_servers,
timeout=timeout
)
def send(self, topic, value, key=None, partition=-1):
return self.client.produce(
topic,
value,
key=key,
partition=partition
)
The consumer group implementation is more interesting because it needs to maintain a heartbeat in the background:
def _heartbeat_loop(self):
interval = max(self.session_timeout / 3.0, 0.5)
while not self._stop.wait(interval):
_, needs_rejoin, _ = self.client.heartbeat(
self.group_id,
self.member_id,
self.generation
)
if needs_rejoin:
self._join()
I tested two consumer groups against the same broker and watched them split the partitions. I then stopped one of them and verified that the remaining consumer eventually reclaimed its partitions after the session timeout.
Running the whole thing
Once the broker had multiple processes and the Python client, running everything manually became annoying. So I added Docker Compose.
The cluster contains three broker containers connected to the same Docker network. Instead of hardcoding container IP addresses, the brokers use Docker Compose service names for discovery.
The project also has integration tests that build the real Go binary and start actual broker processes. That means the tests can check things like:
- TCP communication
- persistence across restarts
- consumer group assignment
- rebalancing
- multiple brokers
- replication
- leader failover
- client redirects
Some tests even kill a broker process to simulate failure.
What I learned
The implementation was useful, but the things I learned from the failures were probably more valuable.
1. Concurrency bugs need actual concurrency
I designed the locking around separate pieces of state rather than using one global lock. The main areas had their own synchronization:
partition -> partition lock
consumer group -> group lock
offset tracking -> offset lock
I then ran the project with Go's race detector:
go test -race ./...
It came back clean across the project. That doesn't prove that a concurrent system has no bugs, but it gave me much more confidence than simply looking at the locking code.
2. Default assumptions matter
The broker initialization bug was a good example. "Assume a peer is dead until proven alive" sounded like the safer choice. In this case, it produced incorrect leadership decisions during startup.
The important part was not that my first choice was wrong. It was realizing that the choice itself affects system behavior.
3. Simplifying is fine if you know what you're giving up
kafka-lite is full of deliberate simplifications:
| Real Kafka | kafka-lite |
|---|---|
| Custom binary protocol | Length-prefixed JSON |
| Full group protocol | Broker-side assignment |
| Consensus/coordination | Deterministic assignment |
| Quorum acknowledgments | Leader-only acknowledgment |
| Production-grade recovery | Simpler local recovery |
For a learning project, I feel like this is more than enough.
Where this leaves me
I started this project because I knew Kafka's vocabulary but didn't feel like I understood the system behind it.
After building kafka-lite, I have a much better mental model. I understand why partitions are the unit of ordering. I understand why a partition needs a clear write authority. I understand why consumers need offsets and why changing consumer membership requires coordination. I understand why replication helps with broker failures, and why replication by itself is not enough to guarantee that an acknowledged write survives.
And much more importantly, I had to make the decisions myself. I had to see what happened when those decisions were wrong.
That was the real point of building kafka-lite.
This project gave me a much deeper understanding and a clearer architectural mental model than any documentation ever could.
Top comments (0)