System requirements

Functional:

Functional Requirements:

1. Producers and Consumers:
   • Producers can publish messages to specific topics.
   • Consumers can subscribe to topics to receive messages.
2. Message Delivery:
   • Messages must be delivered to all subscribers of a topic.
   • Support for both durable subscriptions (where messages are retained for offline subscribers) and nondurable subscriptions (messages are delivered only while consumers are online).
3. Topic Management:
   • Ability to create and delete topics dynamically.
   • Handling topic metadata, including configuration for partitions, retention policies, etc.
4. Message Filtering:
   • Support for filtering messages based on certain criteria (properties or headers).

Non-Functional:

1. Scalability:
   • Must handle an increasing number of topics and subscribers without impacting performance.
2. Low Latency:
   • Messages should be delivered with minimal delay to ensure timely processing.
3. Fault Tolerance:
   • The system should be resilient to node failures, ensuring message retention and delivery.
4. Reliability:
   • Messages should not be lost during system failures; the system should guarantee at least once or exactly once delivery semantics.
5. Performance:
   • Should support high throughput, allowing large volumes of messages to be processed efficiently.

Capacity estimation

Message can be max = 1MB

Auxiallry data will be 1KB in size

We should uspport at average 50000 consumers concurrently

We should support at average 20000 producers concurently

Producer averages 5 messages per second

Producers volume per second is 2000 * 5 = 100000 messages per second

Server can handle 1000 connections, thus we need 20000 / 1000 = 20 app servers

the total amount of data for all messages is (1Mb + 1KB) * 100000 = 10000 GB = 1TB of messages

A server can handle 100GB thus we need 10 servers to hold our data

API design

publish(user_id, topic, message)

subscribe(consumer_id, topic, offset)

subscribe_with_filter(consumer_id, topic, offset, filter_predicate)

delete_topic(user_id, topic)

Database design

Message <| Producer ProducerId

Message : +int PartitionKey

Message : +int ProducerId

Message : +String Topic

Message: +Vec[Byte] Body

Message: +DateTime PostDate

Message: +int Offset

Producer: +int ProducerId PK

Producer: +String Name

We have 2 options where to keep message:

1) In RAM. Pros: faster, Cons: not durable

2) on Disc. Pros: durable, COnst: slower for random access. However because our messages form a queue we can exploit disc optimization where appending to the end of the log is as fast as writing to the RAM thus we choose disc as our storage

High-level design

Request flows

Producer flow

1) Producer connets to the routing layer that give him address of the partition leader

2) producer sends message to the partition leader that routes the message to the appropriate partition holder based on the partition key and topic. Message routing is done using consistent hashing schema.

3) partition leader replicates the messsage across multiple instances of partition replicas and waits for acks depending on delivery semantics

4) producers recevies an ack or nack (or multiple acks)

5) producers deems write as a success

Consumer flow

1) Consumer ocnnects to the routing layer to get the address of the partition leader

2) Consumer is routed to the appropriated partition and sends desired offset from which he wants to read messages

If we have a pull semantic the consumer pulls any messages that are >= than the offset he specifies, if there is nothing hte consumer waits for a signal for new messages with a timeout

If we have a push semantic the consumer receives any new messages from the producer and checks if they match the offset he wants

3) After reading the message consumer writes back to the partition holder the latest offset of messages it has consumed.

4) Consumer gets write ack and deemed reading as a success

Message replication flow:

1) Partition leader receives a new message

2) Partition leader finds a set of replicas by hashing partition key and walking through the consistent hashing ring to find a set of replicas that hold a topic

3) Message write is send to a set of replicas

4) When the majority of replicas answer with an ack the write is deemed to be acked

Message delivery semantics:

1) When user want at most once we as a producer wait only for 0 akcs when deeming the message as a successfully written. If the message fails to be written the producers doesn't retry. The consumer deems read as success when reading the message imemdiately. It doesn't matter if consumed offset is written back or not.

2) When a user wants at least once we wait for 1 ack or for all acks when writing a message. When consuming hte message we ack success only when the offset is written back to the partition holder. If the producer crashes before offset is written the message will be reconsumed. If the producer crashes after the message won't be resend

Producer rebalalnce:

1) Partition replica can detect if producer is down by receiving a constant heartbeats from a consumer

2) If one is down we can send message to others to stop consuming from the replica and to switch to a new one

Detailed component design

Message partition node

1) Every partiion holder is a shard that contains a set of particular topics

2) Every partition holder is replicated multiple times (including partition leader) for durability and to potentially scale read throughput

3) Messages on disc are structured as a sequential log

Coordinator service:

1) Consists of a zookepere multi node system

2) Backed by metadata storage that contains partitions addresses, sizes etc

3) Zookeper provied quorum mechanism to elect new partition leader in case previous one goes down. The use of zookeper is important because it provides a linearizable storage for our leader infromation, that will prevent eelcting 2 leaders at the same time.

Trade offs/Tech choices

Data storage: we use disc as our message storage exclusively because our messages are structured as an append only log which has the best performance on disc + added durability

We expect to have more consumers than producers and thus we choose pull model for message delivery. Push model has the advantage of regulating the consumtion rate from the sonsumer side vs producer side and because we expect our consumption clients to have irregular consumption rate we choose push model.

We opt out of using RAM for messages in favor of disc

To make sure that we can reliably elect a new leader in case original partiion leader is down we use a quorum based log provided by zookeper where all the connected nodes can vote on the new leader depending on who has the latest message log.

Failure scenarios/bottlenecks

If leader coordinator craehs other notes will vote for an ew leader and elect one eventually through zookepere. If parttionholder crashes when performing a write we can continue replicating to existing replicase. When the node will go back up it's state can be rebuilt using previous working replicas. Depending on delivery semantics we can have different latency for writes: at most once, at least once, exactly once (in the order of increasing latency). Depending on our replication strategy (ack = 0,1 or ack all) we can have a diffrent write latency (order in the order of increasing laency). When using push/pull model we can either overwhelm consumers (push model) or starve producers (pull model) in case our power for producing/consuming doesn't match.

Future improvements

Partition otpics further if a partition holder gets overwhelmed.

Measure consumption/prudction rate to adaptively scale producers/consumers by adding timeuts or buffers ofsize