Creating Topics and Partition Storage

Introduction to the Lesson

Today, we'll build the component that brings this design to life: the topic manager. When a client requests to create a new topic (e.g., "user_events" with 4 partitions), something needs to handle that request. It must create the internal data structures for the topic and, most importantly, instantiate the underlying storage for each of those partitions.

Your Goal for Today: Your learning outcome is to implement topic creation logic, including the creation of underlying storage for each partition.

We will design a central Broker component to manage topics and implement a CreateTopic method. This process will introduce a powerful Go pattern for dynamically creating dependencies, solidifying the principles of interface-driven design you've been working with.

1. Visualizing Topics and Partitions

Before we write any code, let's solidify our mental model. A topic is a logical channel for messages, but the unit of storage and parallelism is the partition. Each partition is an independent, ordered commit log.

When we create a topic, we are really creating one or more of these independent logs.

Our goal is to create the Go structs that represent this structure and the logic to build it on demand.

2. Structuring the Broker

We need a central component that will manage the lifecycle of all topics. Let's call it Broker. This struct will be the heart of our application, holding the state of all active topics.

A key requirement is that topic creation must be safe for concurrent access. Multiple clients might try to create topics at the same time. The standard Go tool for protecting shared data like this is the sync.RWMutex.

Let's define our core data structures in a new file, perhaps internal/broker/broker.go:

package broker

import (
	"sync"
	
	"github.com/your-username/your-project/internal/log" // Your CommitLog package
)

// Partition represents a single partition of a topic.
type Partition struct {
	ID  string
	Log log.CommitLog // Each partition has its own commit log.
}

// Topic represents a collection of partitions.
type Topic struct {
	Name       string
	Partitions []*Partition
}

// Broker is the central controller that manages topics.
type Broker struct {
	mu     sync.RWMutex
	topics map[string]*Topic // Map from topic name to Topic struct
    
    // We'll add more to this struct shortly.
}

func NewBroker() *Broker {
	return &Broker{
		topics: make(map[string]*Topic),
	}
}

This structure directly mirrors the diagrams. The Broker holds a map of Topics. Each Topic contains a slice of Partitions, and each Partition embeds the CommitLog interface we've worked so hard on.

3. The Challenge: Dynamic Storage Creation

Here's the central design problem for this lesson: If a request comes in to create a topic with 4 partitions, how do we instantiate four new, independent CommitLog instances?

We can't just pass a single, pre-made log instance to the Broker. That would mean all partitions share the same underlying file or database connection, which completely breaks the model.

We need a way to tell the Broker, "Here is a recipe for creating logs. When you need a new one, use this recipe." In software design, this is a classic use case for the Factory Pattern. In Go, we can implement this elegantly with a higher-order function.

The video you're about to watch explores this exact problem and arrives at this powerful solution. The presenter's thought process is as valuable as the final code.

4. Implementing the Log Factory

Inspired by the video, let's formalize the factory function pattern for our CommitLog.

1. Define the Factory Type: We'll define a function type that represents the "recipe" for creating a CommitLog.

   // internal/log/factory.go
   package log
   
   // LogFactory is a function that creates a new CommitLog for a given partition ID.
   type LogFactory func(partitionID string) (CommitLog, error)
   

2. Update the Broker: The Broker will hold an instance of this factory. This is a form of dependency injection; we are injecting the ability to create logs.

   // internal/broker/broker.go (updated)
   type Broker struct {
       mu         sync.RWMutex
       topics     map[string]*Topic
       logFactory log.LogFactory // The "recipe" for creating logs
   }
   
   type BrokerConfig struct {
       LogFactory log.LogFactory
   }
   
   func NewBroker(config BrokerConfig) *Broker {
       return &Broker{
           topics:     make(map[string]*Topic),
           logFactory: config.LogFactory,
       }
   }
   

3. Create Concrete Factories: Now, in our main.go or setup code, we can create concrete factory functions for each of our storage types.

   // Example for FileLog (in cmd/broker/main.go)
   func newFileLogFactory(baseDir string) log.LogFactory {
       return func(partitionID string) (log.CommitLog, error) {
           logDir := filepath.Join(baseDir, partitionID)
           // Use the LogConfig from your previous lessons
           cfg := log.Config{ ... } 
           return log.NewLog(logDir, cfg)
       }
   }
   
   // Example for PostgresLog (in cmd/broker/main.go)
   func newPostgresLogFactory(pool *pgxpool.Pool) log.LogFactory {
       return func(partitionID string) (log.CommitLog, error) {
           return log.NewPostgresLog(pool, partitionID)
       }
   }
   

   This pattern is incredibly powerful. The Broker only knows about the LogFactory interface. It has no idea if it's creating logs on the filesystem or in PostgreSQL.

5. Implementing CreateTopic

Now we have all the pieces to implement the core logic.

// internal/broker/broker.go

import (
	"fmt"
	// ... other imports
)

func (b *Broker) CreateTopic(name string, numPartitions int) error {
	// Acquire a full write lock to safely modify the topics map.
	b.mu.Lock()
	defer b.mu.Unlock()

	if _, exists := b.topics[name]; exists {
		return fmt.Errorf("topic '%s' already exists", name)
	}

	// Create the topic and its partitions.
	newTopic := &Topic{
		Name:       name,
		Partitions: make([]*Partition, numPartitions),
	}

	for i := 0; i < numPartitions; i++ {
		partitionID := fmt.Sprintf("%s_%d", name, i)

		// Use the factory to create a new, independent log for this partition.
		commitLog, err := b.logFactory(partitionID)
		if err != nil {
			// In a real system, we'd need to clean up logs created so far.
			// For now, we'll just return the error.
			return fmt.Errorf("failed to create log for partition %s: %w", partitionID, err)
		}

		newTopic.Partitions[i] = &Partition{
			ID:  partitionID,
			Log: commitLog,
		}
	}

	// Add the newly created topic to our central map.
	b.topics[name] = newTopic
	
	fmt.Printf("Created topic '%s' with %d partitions\n", name, numPartitions)
	return nil
}

This function is the culmination of our work. It uses the mutex for safety, checks for existing topics, and then iterates to create the requested number of partitions. In each iteration, it calls the logFactory to dynamically provision the underlying storage, whatever that may be.

To see how this fits into the larger picture, the MonKafka article (ID: LINK) shows how a TCP server dispatches requests. Our CreateTopic function is exactly what would be called by the CreateTopicKey handler in its APIDispatcher map. A client command like the one shown in the Go-Kafka repository (create-topic --partitions 4) is what ultimately triggers this entire flow.

Conclusion

Fantastic work today. You've built the machinery at the core of the broker for managing topic lifecycles. This lesson connected the abstract CommitLog interface to the concrete reality of provisioning storage for new topics.

Key Takeaways:

• Modeling in Go: You've translated the conceptual hierarchy of topics and partitions into concrete Go structs.
• Concurrency Control: You used sync.RWMutex to protect the broker's shared state, a fundamental pattern in concurrent Go programming.
• The Factory Function Pattern: You implemented a LogFactory to decouple the Broker from the specific type of storage it creates. This provides immense flexibility and is a testament to your goal of mastering advanced Go patterns.

Next Up

Now that we can create topics, the logical next step is to handle their destruction. In the next lesson, we will implement the DeleteTopic logic. This will involve removing the topic from the broker's state and, importantly, invoking the Close() and Remove() methods on each partition's CommitLog to clean up the underlying storage resources.