Reading from a Partition's Log

Introduction to the Lesson

Today, we will build the fundamental retrieval mechanism for our broker. Your learning outcome is to implement the core logic to read a sequence of messages from a partition's log given a starting offset.

This is the first and most crucial step in building a consumer. Before our broker can send messages to a client over the network, it must first be able to read them from its own disk storage.

In a system like Kafka, a consumer initiates this process with a "Fetch Request." This request tells the broker exactly which topic, which partition, and which message offset to start reading from.

To achieve this, we will:

1. Define a stable, binary on-disk format for our messages.
2. Update our Append logic to write messages in this new format.
3. Implement a Read method that can scan the log file and retrieve all messages from a given starting offset.

This lesson will dive into file I/O and binary data manipulation, foundational skills for any systems programmer.

1. Designing a Robust On-Disk Format

To read data reliably, we must first write it in a predictable format. Simply concatenating message payloads isn't enough, as we wouldn't know where one message ends and the next begins.

We will adopt a common and effective format used in many log-based systems: a length-prefixed record. Each message we store on disk will be a record with the following structure:

When written to the log file, these records will be laid out sequentially:

[Record 0: Offset, Size, Value][Record 1: Offset, Size, Value][Record 2: ...]

This structure is powerful because it allows us to scan the file efficiently. By reading the fixed-size header (Offset + Size = 12 bytes), we know exactly how many bytes to read for the message's value, which then positions us perfectly to read the header of the next record.

To handle the conversion between Go structs and these binary representations, we'll use the standard library's encoding/binary package.

2. Updating the CommitLog to Write Records

Before we can read, we must update our writing logic. In our previous work, the CommitLog struct was responsible for appending raw []byte payloads. We will now upgrade it to write records in the format defined above.

First, let's define a Record struct in Go that mirrors our on-disk structure. You can create this in a new file, internal/broker/record.go.

// internal/broker/record.go
package broker

// Record represents a single message stored in the commit log.
type Record struct {
	Offset uint64
	Value  []byte
}

Next, let's modify the CommitLog struct in internal/broker/commitlog.go. We need to add fields to track the next offset and a mutex to protect concurrent appends.

// internal/broker/commitlog.go
package broker

import (
	"encoding/binary"
	"os"
	"sync"
)

type CommitLog struct {
	path string
	file *os.File
	mu   sync.Mutex
	// The next offset to be assigned to a new record.
	nextOffset uint64
}

// NewCommitLog ... (keep the existing constructor but initialize the new fields)
func NewCommitLog(path string) (*CommitLog, error) {
	// ... (existing code to open/create the file)
	// For now, we assume we start from scratch. We'll handle re-loading state later.
	return &CommitLog{
		path:       path,
		file:       f,
		nextOffset: 0,
	}, nil
}

Now, let's refactor the Append method. Instead of just writing the payload, it will now construct and write the full [Offset][Size][Value] record.

// In internal/broker/commitlog.go

const (
	offsetSize = 8
	lengthSize = 4
	headerSize = offsetSize + lengthSize
)

// Append writes a new record to the commit log.
func (l *CommitLog) Append(value []byte) (uint64, error) {
	l.mu.Lock()
	defer l.mu.Unlock()

	offset := l.nextOffset

	// 1. Create the 12-byte header.
	header := make([]byte, headerSize)
	binary.BigEndian.PutUint64(header[0:offsetSize], offset)
	binary.BigEndian.PutUint32(header[offsetSize:headerSize], uint32(len(value)))

	// 2. Write the header.
	_, err := l.file.Write(header)
	if err != nil {
		return 0, err
	}

	// 3. Write the payload.
	_, err = l.file.Write(value)
	if err != nil {
		// In a real system, you'd need to handle this failure,
		// possibly by truncating the file to remove the partial write.
		return 0, err
	}

	// Ensure data is written to stable storage.
	err = l.file.Sync()
	if err != nil {
		return 0, err
	}

	// 4. Increment the offset for the next message.
	l.nextOffset++

	return offset, nil
}

With this change, our log file is no longer just raw data; it's a structured sequence of self-describing records.

3. Implementing the Core Read Logic

Now for the main event: implementing the Read method. This method will take a startOffset and return a slice of all records from that offset to the end of the log.

The logic will be a loop that:

1. Seeks to the beginning of the file.
2. Reads a record header.
3. Parses the offset and size from the header.
4. If the record's offset is less than startOffset, it skips the payload and continues to the next record.
5. If the record's offset is greater than or equal to startOffset, it reads the payload and appends the complete record to our results.
6. The loop terminates upon reaching the end of the file (io.EOF).

This "scan from the beginning" approach is simple and correct. For a high-performance system like Kafka, this scan is optimized by using a separate index file that maps logical offsets to physical byte positions in the log file, but for now, a linear scan is a perfect way to implement the core logic.

Let's add the Read method to CommitLog.

// In internal/broker/commitlog.go

import (
	// ... other imports
	"bufio"
	"errors"
	"io"
)

// Read reads records from the log starting from a given offset.
func (l *CommitLog) Read(startOffset uint64) ([]Record, error) {
	// We use a separate file handle for reading to avoid interfering
	// with the main file handle used for appending.
	file, err := os.Open(l.path)
	if err != nil {
		return nil, err
	}
	defer file.Close()

	// It's important to seek to the beginning before starting the scan.
	_, err = file.Seek(0, io.SeekStart)
	if err != nil {
		return nil, err
	}

	reader := bufio.NewReader(file)
	var records []Record

	for {
		header := make([]byte, headerSize)
		// io.ReadFull ensures we read exactly headerSize bytes or return an error.
		_, err := io.ReadFull(reader, header)
		if err != nil {
			// A clean EOF means we've read all full records.
			if errors.Is(err, io.EOF) {
				break
			}
			return nil, err
		}

		offset := binary.BigEndian.Uint64(header[0:offsetSize])
		size := binary.BigEndian.Uint32(header[offsetSize:headerSize])

		payload := make([]byte, size)
		_, err = io.ReadFull(reader, payload)
		if err != nil {
			// An EOF here indicates a corrupted record (header without a full payload).
			if errors.Is(err, io.EOF) {
				return nil, io.ErrUnexpectedEOF
			}
			return nil, err
		}

		// If the record's offset is what we're looking for (or later), add it.
		if offset >= startOffset {
			records = append(records, Record{
				Offset: offset,
				Value:  payload,
			})
		}
	}

	return records, nil
}

This implementation mirrors the Stream functionality you might see in libraries designed for this purpose. For example, the immuta append-only log library provides a similar abstraction.

Conclusion

Fantastic work! You have now built the read path for our broker. We have a well-defined on-disk format and a Read method on our CommitLog that can retrieve a sequence of messages from any valid starting point.

Key Takeaways:

• A structured on-disk format (like length-prefixing) is essential for reliable reading.
• The encoding/binary package is the standard tool for converting between Go's primitive types and their byte representations for storage or network transmission.
• The core logic for reading from a log involves scanning records and comparing their logical offsets to the desired starting offset.
• Using io.ReadFull is crucial for ensuring you read the exact number of bytes you expect, which helps detect corrupted data or incomplete writes.

Next Up

We now have the two fundamental storage operations: Append and Read. However, this logic is still buried within the CommitLog struct. In the next lesson, we will expose this functionality over the network by integrating the partition-reading logic into our TCP server. This will allow a remote client to send a fetch request and receive a stream of messages in response, bringing our consumer API to life.