Building a Kafka-style commit log from scratch.
Recently I was learning how Kafka works internally. To understand its append-only log design, I built a simplified commit-log system from scratch in Go.
Some people might think that Kafka is a message queue and not related to logs. But in reality Kafka is an append-only log system. I’ve discussed this more in this post.
In this post we will implement the log system discussed in the previous post from scratch.
A Quick Recap
A commit log is an append-only and immutable sequence of records. Commit logs are used in several software applications, including databases, message queues, event sourcing in microservices, etc.
It acts like a source of truth of actions performed in a system.
For example, Kafka stores each partition as an append-only log on disk. In this post, we will build a simplified commit-log storage layer inspired by that design.
Store File
Inside our commit log, there is a file with the .store extension. This file stores the data written to the commit log using a simple length-prefixed format.
The first 8 bytes of each record store the message length. If I want to store Sushant, its length is 7, so the first 8 bytes store the number 7 and the next 7 bytes store the actual message bytes.
Offset
Every message in this commit log gets a sequential identifier called an offset. The offset helps us locate the record through the index. The first message gets offset 0, the second gets 1, and so on.
Index File
An offset alone is not enough to locate the message in the store file because we still need its byte position. The index file stores a mapping from offset to position in the store file.
Reading Path
When a read request arrives, we first ask the index for the position of the requested offset. We then read 8 bytes from that position to get the message length and read that many bytes for the message body.
Enough Theory. Let’s Code
store.go
1package log
2
3import (
4 "bufio"
5 "encoding/binary"
6 "os"
7 "sync"
8)
9
10var enc = binary.BigEndian
11
12const lenWidth = 8
13
14type store struct {
15 file *os.File
16 mu sync.Mutex
17 buf *bufio.Writer
18 size uint64
19}
20
21func newStore(f *os.File) (*store, error) {
22 fi, err := os.Stat(f.Name())
23 if err != nil {
24 return nil, err
25 }
26
27 size := uint64(fi.Size())
28
29 return &store{
30 file: f,
31 size: size,
32 buf: bufio.NewWriter(f),
33 }, nil
34}
35
36func (s *store) Append(data []byte) (n uint64, pos uint64, err error) {
37 s.mu.Lock()
38 defer s.mu.Unlock()
39
40 // Position at which current data is being stored.
41 pos = s.size
42
43 // This will write length of data in 8 bytes.
44 if err = binary.Write(s.buf, enc, uint64(len(data))); err != nil {
45 return 0, 0, err
46 }
47
48 w, err := s.buf.Write(data)
49 if err != nil {
50 return 0, 0, err
51 }
52
53 // w is the total length we have written for "data"
54 w += lenWidth
55 s.size += uint64(w)
56
57 return uint64(w), pos, nil
58}
59
60func (s *store) Read(pos uint64) ([]byte, error) {
61 s.mu.Lock()
62 defer s.mu.Unlock()
63
64 if err := s.buf.Flush(); err != nil {
65 return nil, err
66 }
67
68 size := make([]byte, lenWidth)
69 if _, err := s.file.ReadAt(size, int64(pos)); err != nil {
70 return nil, err
71 }
72
73 b := make([]byte, enc.Uint64(size))
74 if _, err := s.file.ReadAt(b, int64(pos+lenWidth)); err != nil {
75 return nil, err
76 }
77
78 return b, nil
79}
80
81func (s *store) ReadAt(p []byte, off int64) (int, error) {
82 s.mu.Lock()
83 defer s.mu.Unlock()
84 if err := s.buf.Flush(); err != nil {
85 return 0, err
86 }
87 return s.file.ReadAt(p, off)
88}
89
90func (s *store) Close() error {
91 s.mu.Lock()
92 defer s.mu.Unlock()
93
94 err := s.buf.Flush()
95 if err != nil {
96 return err
97 }
98
99 return s.file.Close()
100}
Let’s understand what’s going on.
1var enc = binary.BigEndian
This line is not just syntax, it’s important:
We are deciding how numbers are stored in bytes. When we write length (like 7 for “Sushant”), it is not stored as “7” — it is stored as 8 bytes in big-endian format. There are two common byte orders: big-endian and little-endian. I chose big-endian for the on-disk format. Network byte order is traditionally big-endian, but for a local file format either byte order can be valid as long as the format is defined consistently.
If you want to understand more about endianness, here is a great article for you.
1const lenWidth = 8
We always use 8 bytes to store length. So even if data length is 5, we still reserve 8 bytes.
This makes reading predictable:
- First read 8 bytes → get length
- Then read next N bytes → actual data
Important parts:
- file → actual file on disk
- buf → buffered writer (performance boost)
- size → current file size (very important for offsets)
- mu → lock to avoid race conditions
The rest of the code is straightforward. Append() writes the length and data, then returns the number of bytes written, the starting position, and any error.
Read() first flushes the buffered writer so that data written through bufio.Writer is visible to the file descriptor. Flush() does not guarantee durable storage; File.Sync() is the relevant operation when stronger durability is required. We then read the record length and the corresponding number of message bytes.
index.go
1/*
2 Data structure for the index:
3 | 4 bytes store offset | 8 bytes store position |
4*/
5
6package log
7
8import (
9 "io"
10 "os"
11
12 "github.com/tysonmote/gommap"
13)
14
15var (
16 offWidth uint64 = 4 // bytes
17 posWidth uint64 = 8 // bytes
18 completeWidth = offWidth + posWidth
19)
20
21type Config struct {
22 Segment struct {
23 MaxStoreBytes uint64
24 MaxIndexBytes uint64
25 InitialOffset uint64
26 }
27}
28
29type index struct {
30 file *os.File
31 mmap gommap.MMap
32 size uint64
33}
34
35func newIndex(f *os.File, config Config) (*index, error) {
36 idx := &index{
37 file: f,
38 }
39
40 fi, err := os.Stat(f.Name())
41 if err != nil {
42 return nil, err
43 }
44
45 idx.size = uint64(fi.Size())
46
47 /*
48 Truncating because later we are mapping this file in memory for faster access.
49 Truncating won't be possible once file is mapped to memory.
50 */
51 if err := os.Truncate(f.Name(), int64(config.Segment.MaxIndexBytes)); err != nil {
52 return nil, err
53 }
54
55 m_map, err := gommap.Map(
56 idx.file.Fd(),
57 gommap.PROT_READ|gommap.PROT_WRITE,
58 gommap.MAP_SHARED,
59 )
60 if err != nil {
61 return nil, err
62 }
63
64 idx.mmap = m_map
65
66 return idx, nil
67}
68
69func (i *index) Close() error {
70 // This asks the OS to synchronize the memory-mapped changes with the file.
71 if err := i.mmap.Sync(gommap.MS_SYNC); err != nil {
72 return err
73 }
74
75 // Even after mmap sync, the OS may still have data in cache.
76 // File.Sync asks the OS to synchronize the file with persistent storage.
77 if err := i.file.Sync(); err != nil {
78 return err
79 }
80
81 // We again need to truncate because when opening this file inside newIndex() we truncated it to config.Segment.MaxIndexBytes
82 // this was done to memory map it. But during restart this will cause issue.
83 // If we keep file truncated to config.Segment.MaxIndexBytes it will have some garbage data in file and when we restart this application our next offset will be incorrect.
84 if err := i.file.Truncate(int64(i.size)); err != nil {
85 return err
86 }
87
88 return i.file.Close()
89}
90
91func (i *index) Write(off uint32, pos uint64) error {
92 if uint64(len(i.mmap)) < i.size+completeWidth {
93 return io.EOF
94 }
95
96 enc.PutUint32(i.mmap[i.size:i.size+offWidth], off)
97 enc.PutUint64(i.mmap[i.size+offWidth:i.size+completeWidth], pos)
98
99 i.size += uint64(completeWidth)
100 return nil
101}
102
103func (i *index) Read(off int64) (out uint32, pos uint64, err error) {
104 if i.size == 0 {
105 return 0, 0, io.EOF
106 }
107
108 if off == -1 {
109 out = uint32(i.size/completeWidth) - 1
110 } else {
111 out = uint32(off)
112 }
113
114 pos = uint64(out) * completeWidth
115
116 if i.size < pos+completeWidth {
117 return 0, 0, io.EOF
118 }
119
120 out = enc.Uint32(i.mmap[pos : pos+completeWidth])
121 pos = enc.Uint64(i.mmap[pos+offWidth : pos+completeWidth])
122
123 return
124}
125
126func (i *index) Name() string {
127 return i.file.Name()
128}
Inside our index file, data is stored in the following format:
| 4 Bytes | 8 Bytes |
An offset for every message can be stored in the first 4 bytes, and the next 8 bytes store the actual position of that message in the store file. So a single index entry will be 12 bytes long. As we know that writes are sequential and offsets are auto-incremented, we can easily query the index file.
For example, if we want to know the message with offset 5, we can simply do math:
5 * 12 = 60
So bytes 60 through 71 contain the index entry for offset 5: bytes 60–63 store the offset and bytes 64–71 store the position in the store file.
Segmentation
1type Config struct {
2 Segment struct {
3 MaxStoreBytes uint64
4 MaxIndexBytes uint64
5 InitialOffset uint64
6 }
7}
We can’t store an unlimited amount of data in a single file forever, so we split the log into segments. A segment contains a store file and an index file. We are not learning about segmenting in detail in this post. Just understand that segments are used to split data into multiple files instead of having one big single file.
For example, a segment can be seen as the following:
100000000.store
200000000.index -> Segment 1
3
400010000.store
500010000.index -> Segment 2
Segment 1 stores messages with an offset from 0 to 9999, and segment 2 stores messages from offset 10000 and above.
1type index struct {
2 file *os.File
3 mmap gommap.MMap
4 size uint64
5}
Everything is familiar to you except 1 thing, mmap.
Memory Mapping
It is a technique in which we map any file into virtual memory. When you are writing or reading a file, we need to use system calls, which are slow. But with memory mapping, instead of using slow read/write system calls, the application reads/writes data directly to disk via memory pointers. It’s like treating our file as an in-memory array.
newIndex() This function creates an object of the index struct. There are few important things here to understand. Before memory mapping our file, we are truncating our file to the maximum size allowed. This is done because once we memory-map our file, we can’t truncate it. Truncating is simple. Let’s say our index file is empty and we defined the max size of the index file to 1024 bytes. Truncating will then add 1024 zeros to this file.
Similarly, if a file currently has 200 bytes of data, it will add 824 zeros at the last of the file.
Then we use Gommap. Map the library to memory map our file.
Writing is simple. As we are treating this file as an array via a memory-mapped file, we just use indexes to write content. We wrote the offset in the first 4 bytes and the position in the next 8 bytes.
Reading is also simple. If the offset provided is -1, we start reading from the end of the file; else, we just use our simple multiplication method to read content from the index. Once we get content from the index, we go to the store file to fetch actual data from it.
One important thing to understand is the Close() method. While closing, we are again truncating our file to the variable size of our index. The size variable holds the size of meaningful data in the file. This is important to do. Imagine a scenario where the server restarts and tries to load this file again; it will see its size is maximum because we added all zeros for memory mapping.
So this is how a simple commit log works. I hope you enjoyed this post. I like connecting with new people. I can be found on LinkedIn.