# Design of a Pub-sub System

### First design <a href="#first-design-0" id="first-design-0"></a>

In the previous lesson, we discussed that a producer writes into topics, and consumers subscribe to a topic to read messages from that topic. Since new messages are added at the end of the queue, we can use [distributed messaging queues](https://ylbbak99.gitbook.io/grokking-modern-system-design-interview/distributed-messaging-queue/system-design-the-distributed-messaging-queue) for topics.

The components we’ll need have been listed below:

* **Topic queue**: Each topic will be a distributed messaging queue so we can store the messages sent to us from the producer. A producer will write their messages to that queue.
* **Database**: We’ll use a relational database that will store the subscription details. For example, we need to store which consumer has subscribed to which topic so we can provide the consumers with their desired messages. We’ll use a relational database since our consumer-related data is structured and we want to ensure our data integrity.
* **Message director**: This service will read the message from the topic queue, fetch the consumers from the database, and send the message to the consumer queue.
* **Consumer queue**: The message from the topic queue will be copied to the consumer’s queue so the consumer can read the message. For each consumer, we’ll define a separate distributed queue.
* **Subscriber**: When the consumer requests a subscription to a topic, this service will add an entry into the database.

The consumer will subscribe to a topic, and the system will add the subscriber’s details to the database. The producer will write into the topics, and the message director will read the message from the queue, fetch the details to whom it should add the message, and send it to them. The consumers will consume the message from their queue.

> **Note**: We’ll use fail-over services for the message director and subscriber to guard against failures.

<figure><img src="https://1055613405-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MlQzjDKRh_AaQvZflWz%2Fuploads%2FRYz0V90UWuWb2gEM6cNE%2FScreenshot%202023-09-03%20at%201.03.32%20AM.png?alt=media&#x26;token=56939074-6ac3-4958-bbd2-40f7c6e861b9" alt=""><figcaption></figcaption></figure>

Using the distributed messaging queues makes our design simple. However, the huge number of queues needed is a significant concern. If we have millions of subscribers for thousands of topics, defining and maintaining millions of queues is expensive. Moreover, we’ll copy the same message for a topic in all subscriber queues, which is unnecessary duplication and takes up space.

**Question 1**

Is there a way to avoid maintaining a separate queue for each reader?

In messaging queues, the message disappears after the reader consumes it. So, what if we add a counter for each message? The counter value decrements as a subscriber consumes the message. It does not delete the message until the counter becomes zero. Now, we don’t need to keep a separate queue for each reader.

**Question 2**

What is the problem with the previous approach?

The unread messages can become a bottleneck if we use the conventional queue API. For example, if 9 out of 10 readers have consumed the message present at the start of the queue, then that message won’t be deleted until the tenth consumer has also consumed the message, and the first nine consumers won’t be able to move forward.

We’ll need to change the storage interface so that consumers can independently consume data. Our system will need to keep sufficient metadata and track what information each consumer has consumed and delete a message when the information has been consumed by all consumers. It resembles with reference count mechanism in Linux’s hard link of files.

\---------

### Second design <a href="#second-design-0" id="second-design-0"></a>

Let’s consider another approach to designing a pub-sub system.

#### High-level design <a href="#high-level-design-1" id="high-level-design-1"></a>

At a high level, the pub-sub system will have the following components:

* **Broker**: This server will handle the messages. It will store the messages sent from the producer and allow the consumers to read them.
* **Cluster manager**: We’ll have numerous broker servers to cater to our scalability needs. We need a cluster manager to supervise the broker’s health. It will notify us if a broker fails.
* **Storage**: We’ll use a relational database to store consumer details, such as subscription information and retention period.
* **Consumer manager**: This is responsible for managing the consumers. For example, it will verify if the consumer is authorized to read a message from a certain topic or not.

Besides these components, we also have the following design considerations:

* **Acknowledgment**: An acknowledgment is used to notify the producer that the received message has been stored successfully. The system will wait for an acknowledgment from the consumer if it has successfully consumed the message.
* **Retention time**: The consumers can specify the retention period time of their messages. The default will be seven days, but it is configurable. Some applications like banking applications require the data to be stored for a few weeks as a business requirement, while an analytical application might not need the data after consumption.

<figure><img src="https://1055613405-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MlQzjDKRh_AaQvZflWz%2Fuploads%2FHKiz1iX3XuYlhRYMgcBg%2FScreenshot%202023-09-03%20at%201.04.59%20AM.png?alt=media&#x26;token=40ffd99d-b0a5-4593-9b9a-da18a412a5c8" alt=""><figcaption></figcaption></figure>

Let’s understand the role of each component in detail.

### Broker <a href="#broker-0" id="broker-0"></a>

The broker server is the core component of our pub-sub system. It will handle write and read requests. A broker will have multiple topics where each topic can have multiple **partitions** associated with it. We use partitions to store messages in the local storage for persistence. Consequently, this improves availability. Partitions contain messages encapsulated in **segments**. Segments help identify the start and end of a message using an offset address. Using segments, consumers consume the message of their choice from a partition by reading from a specific offset address. The illustration below depicts the concept that has been described above.

<figure><img src="https://1055613405-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MlQzjDKRh_AaQvZflWz%2Fuploads%2FdP5WJicdoJlftxhTvz0w%2FScreenshot%202023-09-03%20at%201.05.17%20AM.png?alt=media&#x26;token=1defa50d-cebf-43c3-a091-b098ced50b93" alt=""><figcaption></figcaption></figure>

As we know, a **topic** is a persistent sequence of messages stored in the local storage of the broker. Once the data has been added to the topic, it cannot be modified. Reading and writing a message from or to a topic is an I/O task for computers, and scaling such tasks is challenging. This is the reason we split the topics into multiple partitions. The data belonging to a single topic can be present in numerous partitions. For example, let’s assume have Topic A and we allocate three partitions for it. The producers will send their messages to the relevant topic. The messages received will be sent to various partitions on basis of the round-robin algorithm. We’ll use a variation of round-robin: weighted round-robin. The following slides show how the messages are stored in various partitions belonging to a single topic.

![](https://1055613405-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MlQzjDKRh_AaQvZflWz%2Fuploads%2F7hfq2OCfAIZLaD11BY0M%2FScreenshot%202023-09-03%20at%201.09.24%20AM.png?alt=media\&token=d62cdbbe-267e-49f1-ab99-5ea1b4376425)

![](https://1055613405-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MlQzjDKRh_AaQvZflWz%2Fuploads%2FjxEVH8XALBCsdrBN4oAf%2FScreenshot%202023-09-03%20at%201.09.43%20AM.png?alt=media\&token=c8e1beab-dd56-4e6f-8268-a983747e93e8)

![](https://1055613405-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MlQzjDKRh_AaQvZflWz%2Fuploads%2Fup5le5aIt7T7pGsOOM0F%2FScreenshot%202023-09-03%20at%201.10.01%20AM.png?alt=media\&token=abfbb8a2-e923-493d-97b2-836d33c1f589)

![](https://1055613405-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MlQzjDKRh_AaQvZflWz%2Fuploads%2FOKhwsqDn6wcLyaS9lSxZ%2FScreenshot%202023-09-03%20at%201.10.17%20AM.png?alt=media\&token=722c1f80-f72c-4bd3-8b52-2f36f8f208f9)

**Question 1**

Strict ordering ensures that the messages are stored in the order in which they are produced. How can we ensure strict ordering for our messages?

We’ll assign each partition a unique ID, `partition_ID`. The user can provide the `partition_ID` while writing into the system. In this way, the messages will be sent to the specified partition, and the ordering will be strict. Our API call to write into the pub-sub system looks like this:

```txt
write(topic_ID, partition_ID, message)
```

If the user does not provide the `partition_ID`, we’ll use the weighted round-robin algorithm to decide which message has to be sent to which partition.

It might seem strange to give the ability to choose a partition to the client of pub-sub. However, such a facility can be the basis from where clients can get data for some specific time period—for example, getting data from yesterday. For simplicity, we won’t include time-based reading in our design.

**Question 2**

What problems can arise if all partitions are on the same broker?

If the broker fails or dies, all the messages in the partitions will be lost. To avoid this, we need to make sure that the partitions are spread on different brokers.

**Question 3**

Why can’t we use blob stores like S3 to keep messages, instead of the broker’s local storage?

Blob stores like S3 are not optimized for writing and reading short-sized data. If our data is geo-replicated, the above problem is exacerbated.

Therefore, we used the server’s local persistent store with append-based writing. Traditional hard disks are specially tuned to provide good write performance with writing to contiguous tracks or sectors. Reading throughput and latency is also good for contiguous regions of the disk because it allows extensive data caching.

**Question 4**

If we use a round-robin algorithm to send messages to a partition, how does the system know where to look when it is time to read?

Our system will need to keep appropriate metadata persistently. This metadata will keep mappings between the logical index of segment or messages to the server identity or partition identifier.

We’ll discuss consumer manager later in the lesson, which will keep the required information.

\-------------

We’ll allocate the partitions to various brokers in the system. This just means that different partitions of the same topic will be in different brokers. We’ll follow strict ordering in partitions by adding newer content at the end of existing messages.

Consider the slides below. We have various brokers in our system. Each broker has different topics. The topic is divided into multiple partitions.

![](https://1055613405-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MlQzjDKRh_AaQvZflWz%2Fuploads%2F2ongcVhJEcgokIvUZc2k%2FScreenshot%202023-09-03%20at%201.12.17%20AM.png?alt=media\&token=b26f517d-cd53-44f3-bd5d-e8a14aa88b65)

![](https://1055613405-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MlQzjDKRh_AaQvZflWz%2Fuploads%2F5zhjFK7Qz836THC6so0W%2FScreenshot%202023-09-03%20at%201.12.33%20AM.png?alt=media\&token=40e2ae64-43d6-4b81-8ae7-5f844325e9eb)

![](https://1055613405-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MlQzjDKRh_AaQvZflWz%2Fuploads%2FavCkLhmxjlw5E8yc3qWW%2FScreenshot%202023-09-03%20at%201.12.46%20AM.png?alt=media\&token=55be0b7a-98fb-441c-8ce0-d860223e4653)

We discussed that a message will be stored in a segment. We’ll identify each segment using an offset. Since these are immutable records, the readers are independent and they can read messages anywhere from this file using the necessary API functions. The following slides show the segment level detail.

![](https://1055613405-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MlQzjDKRh_AaQvZflWz%2Fuploads%2FURS23BnZoLAsErD4fHlU%2FScreenshot%202023-09-03%20at%201.12.58%20AM.png?alt=media\&token=ed78b62d-53e7-451c-a5bb-b369b27adde4)

![](https://1055613405-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MlQzjDKRh_AaQvZflWz%2Fuploads%2F7XhXvpl2CbQyMAinr9tO%2FScreenshot%202023-09-03%20at%201.13.14%20AM.png?alt=media\&token=e1feb20d-0f4f-4ece-b8d4-69617bc08c5f)

The broker solved the problems that our first design had. We avoided a large number of queues by partitioning the topic. We introduced parallelism using partitions that avoided bottlenecks while consuming the message.

### Cluster manager <a href="#cluster-manager-0" id="cluster-manager-0"></a>

We’ll have multiple brokers in our cluster. The cluster manager will perform the following tasks:

* **Broker and topics registry**: This stores the list of topics for each broker.
* **Manage replication**: The cluster manager manages replication by using the leader-follower approach. One of the brokers is the leader. If it fails, the manager decides who the next leader is. In case the follower fails, it adds a new broker and makes sure to turn it into an updated follower. It updates the metadata accordingly. We’ll keep three replicas of each partition on different brokers.

<figure><img src="https://1055613405-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MlQzjDKRh_AaQvZflWz%2Fuploads%2FvXuDYUASSC98lMPmJu8W%2FScreenshot%202023-09-03%20at%201.13.23%20AM.png?alt=media&#x26;token=367e53cf-4a00-4918-862d-e0259ddbb9aa" alt=""><figcaption></figcaption></figure>

### Consumer manager <a href="#consumer-manager-0" id="consumer-manager-0"></a>

The consumer manager will manage the consumers. It has the following responsibilities:

* **Verify the consumer**: The manager will fetch the data from the database and verify if the consumer is allowed to read a certain message. For example, if the consumer has subscribed to Topic A (but not to Topic B), then it should not be allowed to read from Topic B. The consumer manager verifies the consumer’s request.
* **Retention time management**: The manager will also verify if the consumer is allowed to read the specific message or not. If, according to its retention time, the message should be inaccessible to the consumer, then it will not allow the consumer to read the message.
* **Message receiving options management**: There are two methods for consumers to get data. The first is that our system pushes the data to its consumers. This method may result in overloading the consumers with continuous messages. Another approach is for consumers to request the system to read data from a specific topic. The drawback is that a few consumers might want to know about a message as soon as it is published, but we do not support this function.

  Therefore, we’ll support both techniques. Each consumer will inform the broker that it wants the data to be pushed automatically or it needs the data to read itself. We can avoid overloading the consumer and also provide liberty to the consumer. We’ll save this information in the relational database along with other consumer details.
* **Allow multiple reads**: The consumer manager stores the offset information of each consumer. We’ll use a key-value to store offset information against each consumer. It allows fast fetching and increases the availability of the consumers. If Consumer 1 has read from offset 0 and has sent the acknowledgment, we’ll store it. So, when the consumer wants to read again, we can provide the next offset to the reader for reading the message.

### Finalized design <a href="#finalized-design-1" id="finalized-design-1"></a>

The finalized design of our pub-sub system is shown below.

<figure><img src="https://1055613405-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MlQzjDKRh_AaQvZflWz%2Fuploads%2FKDo5WQGSuJ8e7mWtqLFK%2FScreenshot%202023-09-03%20at%201.13.54%20AM.png?alt=media&#x26;token=e4ee82f1-476d-4a87-84d2-9e9d7eef59f9" alt=""><figcaption></figcaption></figure>

### Conclusion <a href="#conclusion-0" id="conclusion-0"></a>

We saw two designs of pub-sub, using queues and using our custom storage optimized for writing and reading small-sized data.

There are numerous use cases of a pub-sub. Due to decoupling between producers and consumers, the system can scale dynamically, and the failures are well-contained. Additionally, due to proper accounting of data consumption, the pub-sub is a system of choice for a large-scale system that produces enormous data. We can determine precisely which data is needed and not needed.
