System requirements

Stakeholders: Producer, Consumer

Functional:

• Producer send messages through the system, the message should be delivered asynchronously. The message should be guaranteed to be delivered
• Consumer receives the message through the system
• The message delivery should be agreed by both parties. Producer and consumer should not send/receive message to the undesired service.

Non-Functional:

• Scalability: Handle large volume of data
• Fault tolerance: Single service failure should not disrupt the overall system.
• Low latency
• Consistency: The state of the system is consistent across instances
• Security: The unauthorized accesses should not be allowed
• Data Persistency: The messages should not be lost.

Capacity estimation

Assuming that 100000 messages per second, each message size is 1 KB.

Throughput: 100 MB per second

Data storage: 60 TB per week.

API design

Publisher:

• POST /publish_message
  • Request: int topic_id, String message_content
  • Response: Status, int message_id
• GET /message_status
  • Request: int message_id
  • Response: Status, String message_status, timestamp created_at, int num_receipients
• POST /create_topic
  • Request: int publisher_id, String topic_name
  • Response: Status, int topic_id

Consumer:

• POST /subscribe
  • Request: int consumer_id, int topic_id
  • Response: Status, Datetime subscribe_at
• POST /unsubscribe
  • Request: int consumer_id, int topic_id
  • Response: Status, Datetime unsubscribe_at
• GET /message
  • Request: int consumer_id, int topic_id
  • Response: Status, int message_id, String message_content

Database design

Metadata: Zookeeper

• Topic
  • PrimaryKey int topic_id
  • String topic_name
  • DateTime created_at
  • int rule_id
  • int pattern_id
• User
  • PrimaryKey int user_id
  • String user_name
  • DateTime created_at

Message: write ahead log

• PrimaryKey int message_id
• DateTime created_at
• ForeignKey int topic_id

Publish: zookeeper

• PrimaryKey int publisher_id
• ForeignKey int topic_id
• ForeignKey int publisher_id
• DateTime created_at

Subscribe: zookeeper

• PrimaryKey int subscription_id
• ForeignKey int topic_id
• ForeignKey int subscriber_id
• DateTime created_at

Message deliver tracking: DB

• PrimaryKey int delivery_id
• ForeignKey int message_id
• ForeignKey consumer_id

High-level design

Key components:

• Poducer: produces message to topics
• Consumer: subscribe to topic and consumes messages
• Message storage: Keeps the messages. The storage is used for two purposes, so it can be split into 2 separate storage system: for low latency message delivery; for tracking and for record
• Metadata storage: configuration and properties of topics. Since we designed the distributed system, so the metadata needs to be consistent over all the replicas, so keeping in Zookeeper is more suitable.
• Coordination service: leader election, keeps replication configures ACK
• Control plane: Receives configuration requests and handle them.
• Brokers: sit in the middle of producer and consumer. Keeps topic and write records

Request flows

Detailed component design

• Scaling strategy of each component
  • Brokers are designed as a distributed system, with each broker managing a partition of a topic. To enhance fault tolerance, multiple replicas of each partition are distributed across different brokers. Each broker may host several replicas from various partitions. For each partition's replica, a designated leader coordinates the message flow, while multiple follower replicas synchronize with the leader. The leader is responsible for receiving newly produced messages and ensuring they are synchronized with its followers. Consumers can issue read requests to any broker housing the corresponding replica. To ensure that 1 service going down does not make message loss, it is necessary to use a quorum acknowledgement for reading and writing to the brokers. For the load balancing, we can use the consistent hashing algorithm to make sure that message and consumption can be routed to the same broker node.
  • Consumers can be scaled separately
• Message delivery pattern
  • For the topics with a requirement with low latency, we can implement a push model. Otherwise, we can use a pull model, that consumer periodically pulls batches of messages from the brokers.
  • At least once: when producer produces the message to the queue, we need to ensure the acknowledgement is met, so that if it is not met, we need to retry from producers.
  • At most once: Does not retry when producer fails; When the message is consumed, look at the consumption first.
• Message storage
  • The message storage can be a part of the broker. Brokers write messages continuously to memory, and once the size reached, write the segment to disk.
  • Once the data retention configuration is reached, write the messages to another storage service to do analysis.

Trade offs/Tech choices

• Pull model vs push model: If we ensure latency first, then we choose push model; if we ensure throughput first, then we choose pull model
• quorum ACK: If we want to ensure lower latency, we can implement lower ACK; if we ensure consistency, we can implement higher ACK with higher latency
• complexity vs maintainability: more rules brings higher maintain cost.

Failure scenarios/bottlenecks

• Broker leader fail: The load balancer in the cluster need to rebalance the old requests to the other nodes. The coordination service uses the leader election algorithm to elect a new leader among the replicas, coordination service brings up a new node and replicate from the new leader.
• Broker follower fail: The coordination service brings up a new follower, sync from the leader.
• consumer fail: auto scaling brings up a new consumer and config based on the existing topic and partitions

Future improvements

What are some future improvements you would make? How would you mitigate the failure scenario(s) you described above?