Implementing Partition Reading in TCP Servers

Introduction to the Lesson

Today, we bridge the gap between our storage layer and the outside world. The learning outcome for this lesson is to integrate the partition-reading logic into the TCP server to handle consumer requests.

This means a client will be able to connect to our broker over the network, ask for messages from a specific topic and partition, and receive them. This is the heart of the "consumer" functionality and a major milestone in our project.

To accomplish this, we will:

1. Define a simple binary protocol for "Fetch" requests and responses.
2. Update our TCP server to act as a request dispatcher, routing incoming connections to the correct handler.
3. Implement a handleFetchRequest function that parses the request, calls our existing CommitLog.Read method, and writes the results back to the client.

This lesson will solidify your understanding of how network services are built on top of storage abstractions, a fundamental pattern in system design.

1. The Fetch Protocol: A Language for Consumers

For a client (a consumer) to retrieve messages, it needs a way to communicate its intent to the broker. We'll define a simple, custom binary protocol for this. Like the length-prefixed record format we designed for disk storage, we need a clear and unambiguous format for network communication.

We'll add a new request type to our broker's vocabulary: the FetchRequest.

Fetch Request Structure

A client will send a byte stream with the following structure:

Fetch Response Structure

When the broker has retrieved the messages, it will send them back in a structured response. Sending the raw data isn't enough; the client needs to know how to parse the stream. We'll send the number of records first, followed by each record, itself length-prefixed.

This "metadata-first" approach is fundamental to reliable streaming over TCP. You can't just io.Copy the data and hope the receiver knows when to stop. You must first tell the receiver how much data to expect.

2. Building a Request Dispatcher

As our broker's capabilities grow, it needs to handle more than one type of request. A clean way to manage this is to create a request dispatcher. The server's connection handler will read just enough data to identify the request type, then pass the connection to a specialized function for full processing.

This pattern is common in broker implementations. The handleConnection function becomes a lightweight router.

Let's see how this looks in practice. We'll update the handleConnection function we sketched out in previous modules.

// In a new file, e.g., internal/server/server.go

const (
    ProduceRequestType uint16 = 1
    FetchRequestType   uint16 = 2
)

func (s *Server) handleConnection(conn net.Conn) {
    defer conn.Close()

    // Read the 2-byte request type
    requestTypeBytes := make([]byte, 2)
    _, err := io.ReadFull(conn, requestTypeBytes)
    if err != nil {
        log.Printf("could not read request type: %v", err)
        return
    }
    requestType := binary.BigEndian.Uint16(requestTypeBytes)

    // Dispatch to the appropriate handler
    switch requestType {
    case ProduceRequestType:
        // s.handleProduceRequest(conn) // We would implement this similarly
        log.Println("Received Produce request (handler not implemented)")
    case FetchRequestType:
        s.handleFetchRequest(conn)
    default:
        log.Printf("received unknown request type: %d", requestType)
    }
}

This dispatcher logic is elegant and extensible. Adding a new feature, like "Create Topic," is as simple as defining a new request type and adding a case to the switch statement.

3. Implementing the Fetch Handler

Now we arrive at the core of today's lesson: the handleFetchRequest function. This is where the network layer meets the storage layer.

This function is responsible for:

1. Parsing the FetchRequest from the connection.
2. Locating the correct CommitLog for the requested partition.
3. Calling commitLog.Read() to retrieve the data from disk.
4. Serializing the records into the FetchResponse format.
5. Writing the response back to the connection.

First, let's assume our Server struct holds the commit logs. For now, we'll keep it simple and manage a single log.

// internal/server/server.go
package server

import (
	"our-kafka/internal/broker"
	// ... other imports
)

type Server struct {
	listener  net.Listener
	commitLog *broker.CommitLog // For simplicity, one log for now
}

func New(listenAddr string) (*Server, error) {
	// ... setup listener ...

	// For this lesson, we hardcode the path to one log file.
	cl, err := broker.NewCommitLog("partition-0.log")
	if err != nil {
		return nil, err
	}

	return &Server{
		listener:  ln,
		commitLog: cl,
	}, nil
}

Now, let's implement the handler.

// internal/server/server.go

func (s *Server) handleFetchRequest(conn net.Conn) {
    // 1. Parse the FetchRequest
    // We already read the 2-byte type code. Now read the rest.

    // Read topic length (uint16)
    topicLenBytes := make([]byte, 2)
    if _, err := io.ReadFull(conn, topicLenBytes); err != nil {
        log.Printf("error reading topic length: %v", err)
        return
    }
    topicLen := binary.BigEndian.Uint16(topicLenBytes)

    // Read topic name, partition ID, and offset
    // Total size = topicLen + 4 (partition) + 8 (offset)
    requestBody := make([]byte, int(topicLen)+4+8)
    if _, err := io.ReadFull(conn, requestBody); err != nil {
        log.Printf("error reading fetch request body: %v", err)
        return
    }

    topic := string(requestBody[0:topicLen])
    // partitionID := binary.BigEndian.Uint32(requestBody[topicLen : topicLen+4])
    startOffset := binary.BigEndian.Uint64(requestBody[topicLen+4 : topicLen+4+8])

    log.Printf("Received fetch request for topic '%s', offset %d", topic, startOffset)
    
    // 2. INTEGRATION POINT: Call the CommitLog's Read method
    // In a real system, we'd use topic and partitionID to find the right log.
    records, err := s.commitLog.Read(startOffset)
    if err != nil {
        log.Printf("error reading from commit log: %v", err)
        // In a real system, you'd send an error response to the client.
        return
    }

    // 3. Serialize the FetchResponse
    responseBuffer := new(bytes.Buffer)

    // Write number of records
    numRecords := uint32(len(records))
    if err := binary.Write(responseBuffer, binary.BigEndian, numRecords); err != nil {
        log.Printf("error writing num records to buffer: %v", err)
        return
    }

    // Write each record (size-prefixed)
    for _, record := range records {
        recordSize := uint32(len(record.Value))
        // Write record size
        if err := binary.Write(responseBuffer, binary.BigEndian, recordSize); err != nil {
            log.Printf("error writing record size to buffer: %v", err)
            return
        }
        // Write record value
        if _, err := responseBuffer.Write(record.Value); err != nil {
            log.Printf("error writing record value to buffer: %v", err)
            return
        }
    }

    // 4. Write the entire response buffer to the connection
    _, err = conn.Write(responseBuffer.Bytes())
    if err != nil {
        log.Printf("failed to write fetch response: %v", err)
    }

    log.Printf("Sent %d records to client", numRecords)
}

This function beautifully illustrates the broker's role as a mediator between network requests and the underlying storage system. The logic from the Sentry article in resource LINK describes this exact flow: the broker looks up the offset, seeks to the byte position, reads the messages, and returns them to the consumer.

Conclusion

Excellent work today! You have successfully wired up the read-path of our message broker. We now have a functioning TCP server that can accept fetch requests from a client, retrieve data from the correct partition log using the logic we built previously, and send a well-formed binary response back over the network.

Key Takeaways:

• Request Dispatching: A robust server uses a dispatcher to route incoming requests to specific handlers based on a type identifier (e.g., an API key or message type code).
• Protocol is King: Defining a clear, unambiguous binary protocol is essential for reliable network communication. Sending size metadata before variable-length data is a critical pattern.
• System Integration: The handleFetchRequest function serves as the glue, translating a network request into a call to the storage layer and then packaging the result for network transmission.

Next Up

Our current consumer model is a simple, synchronous request-response. The client asks, the server answers, and the connection ends. This is inefficient if a consumer wants to continuously listen for new messages. In the next lesson, we will explore a more advanced consumption model. We will learn how to use the select statement with timeouts to multiplex across channel operations without blocking, allowing a consumer handler to efficiently wait for new messages to arrive or handle client timeouts, making our broker far more responsive and powerful.