Implementing Topic Deletion with Storage Cleanup

Introduction to the Lesson

Now that topics can be created, we must address the other side of the lifecycle: deletion. Implementing deletion logic might seem as simple as removing an entry from a map, but in systems that manage persistent resources like files or database tables, it's a critical operation fraught with potential pitfalls. Careless deletion can lead to orphaned resources, data corruption, or system crashes.

Your Goal for Today: Your learning outcome is to implement topic deletion logic, ensuring proper cleanup of associated storage resources.

We will first implement a straightforward deletion method, then analyze its shortcomings. Drawing on professional-grade system design principles, we will then refactor our approach to be more robust and safe for a concurrent environment, solidifying your understanding of the challenges involved in managing stateful systems.

1. The Naive Approach to Deletion

Let's start by implementing the most direct approach to deleting a topic. The Broker holds a map of topics, and each topic's partitions hold a reference to a CommitLog that knows how to clean itself up (via the Close and Remove methods).

The logical flow would be:

1. Find the topic to be deleted.
2. For each of its partitions, close and remove its underlying log.
3. Remove the topic from the broker's central map.

Let's add a DeleteTopic method to our Broker struct in internal/broker/broker.go:

// internal/broker/broker.go

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

func (b *Broker) DeleteTopic(name string) error {
	b.mu.Lock()
	defer b.mu.Unlock()

	topic, exists := b.topics[name]
	if !exists {
		return fmt.Errorf("topic '%s' not found", name)
	}

	// First, clean up all the underlying storage for each partition.
	for _, partition := range topic.Partitions {
		if err := partition.Log.Remove(); err != nil {
			// This is problematic. What if one Remove() fails?
			// We've partially deleted the topic.
			return fmt.Errorf("failed to remove log for partition %s: %w", partition.ID, err)
		}
	}

	// Finally, remove the topic from the in-memory map.
	delete(b.topics, name)
	
	fmt.Printf("Deleted topic '%s'\n", name)
	return nil
}

This code looks plausible. It correctly uses a write lock to ensure exclusive access and calls the necessary cleanup methods on the CommitLog interface. However, this synchronous approach has serious flaws that can cause major problems in real-world systems.

2. The Perils of Synchronous Deletion

In a distributed or even just a highly concurrent system, deleting resources isn't as simple as os.Remove(). What happens if another part of the system is using a resource at the exact moment you delete it? What if one step of your multi-step deletion process fails?

These are fundamental questions in system design. The following article from WarpStream, a company building a Kafka-compatible system on object storage, provides an excellent and highly relevant analysis of this exact problem.

As the article highlights, our naive implementation suffers from two key risks:

1. Resource Leaks: In our code, if the Remove() call for the third partition fails, the method returns an error. The topic remains in the broker's map, but two of its partitions have had their storage deleted. The topic is now in a corrupt, partially-deleted state. Conversely, if we deleted from the map first and then a Remove() failed, the broker would think the topic is gone, but its files or database tables would be left behind forever—an "orphaned resource".
2. Race Conditions: The Broker's lock only protects the topics map. It does not prevent another goroutine (e.g., a network handler processing a client request) from reading from a partition's log. If our DeleteTopic method gets the lock and deletes a log file while another goroutine is actively reading it, that goroutine will crash.

3. A More Robust Deletion Pattern

To solve these problems, we can borrow another concept from the WarpStream article and general system design: separating logical deletion from physical deletion.

• Logical Deletion: Quickly marking the resource as "deleted" in the central metadata store so no new operations can use it. This should be a fast, atomic operation.
• Physical Deletion: The slower, potentially fallible process of cleaning up the underlying resources (files, database tables, etc.).

We can implement a version of this pattern by making our DeleteTopic method perform only the logical deletion while holding the lock, and then delegating the slow, physical cleanup to a background goroutine.

Let's refactor DeleteTopic:

// internal/broker/broker.go

func (b *Broker) DeleteTopic(name string) error {
	b.mu.Lock()
	topic, exists := b.topics[name]
	if !exists {
		b.mu.Unlock() // Release the lock before returning
		return fmt.Errorf("topic '%s' not found", name)
	}

	// 1. Logical Deletion: Remove the topic from the map immediately.
	// No new operations can find this topic anymore.
	delete(b.topics, name)
	
	// 2. Release the lock *before* doing slow I/O.
	// This makes the broker responsive to other requests.
	b.mu.Unlock()

	fmt.Printf("Logically deleted topic '%s', starting background cleanup\n", name)

	// 3. Physical Deletion: Perform cleanup in the background.
	go func(partitionsToClean []*Partition) {
		fmt.Printf("Cleaning up %d partitions for topic '%s'\n", len(partitionsToClean), name)
		for _, partition := range partitionsToClean {
			if err := partition.Log.Remove(); err != nil {
				// In a production system, we would need robust error logging here.
				fmt.Printf("Error cleaning up partition %s: %v\n", partition.ID, err)
			}
		}
		fmt.Printf("Finished cleanup for topic '%s'\n", name)
	}(topic.Partitions) // Pass the partitions slice to the goroutine.

	return nil
}

Analysis of the New Approach

This design is a significant improvement:

• Responsiveness: The broker's critical path (anything requiring the mu lock) is no longer blocked by slow disk I/O from Remove(). The logical deletion is nearly instantaneous.
• Reduced Race Conditions: Once the topic is removed from the map, no new requests can be dispatched to its partitions. While this doesn't solve the "in-flight query" problem for requests that have already started, it dramatically narrows the window for such races. A full solution for in-flight requests would require reference counting or a grace period, which adds significant complexity we can explore later.
• Isolation: We pass the topic.Partitions slice as an argument to the goroutine. This is crucial. It ensures the goroutine works on a self-contained list of partitions and doesn't need to access any shared state from the Broker, avoiding potential data races.

This pattern represents a trade-off. It's much safer than the naive approach, but it's not perfectly fault-tolerant. If the server crashes after the logical deletion but before the background cleanup finishes, the resources will still be orphaned. The professional solution, as hinted at in the WarpStream article's "Delayed Queue" and "Asynchronous Reconciliation" sections, involves persisting the list of "pending deletions" to disk to survive a restart. For our project, this in-memory asynchronous cleanup is an excellent and common intermediate step.

Conclusion

In this lesson, you implemented one of the most deceptively complex parts of a stateful system: resource deletion. You've moved beyond a simple implementation to one that considers concurrency, responsiveness, and failure modes.

Key Takeaways:

• Synchronous, multi-step deletion is brittle and can lead to orphaned resources or race conditions.
• Separating logical deletion (fast, atomic, under a lock) from physical deletion (slow, I/O-bound, in the background) is a robust and common system design pattern.
• Using a goroutine for physical cleanup improves broker responsiveness, but introduces the need to consider failure scenarios like server crashes during cleanup.
• The CommitLog interface allows the Broker to remain blissfully unaware of whether it's deleting files from a disk or dropping tables in PostgreSQL, proving the value of our earlier design choices.

Next Up

We have now implemented the full lifecycle for topics: creation and deletion. The broker can manage its inventory. The next step is to define how clients interact with this inventory. In the upcoming lesson, we will focus on partition assignment logic. When a producer wants to send a message to a topic with multiple partitions, how does it decide which partition to send it to? We'll explore strategies like round-robin and key-based hashing to ensure data is distributed effectively.