Loading...
Estimate the scale of the system. Consider daily active users, read/write ratio, storage requirements, bandwidth, and any relevant QPS calculations...
The 3:1 read-to-write ratio is typical. Three consumer groups (say, analytics, search indexing, and real-time alerting) each read the full stream independently. This is a key advantage of the commit log model, a single write serves many readers without duplicating data. In a traditional message queue, serving three consumers would require either three copies of each message or a complex fanout mechanism. The commit log eliminates this. all consumers read from the same physical log at their own pace.
The replication factor of 3 means every message exists on three separate disks across three different brokers. This is the foundation of durability. any single broker can fail completely and no data is lost because two other copies exist elsewhere. The storage cost of 3x is the price of fault tolerance.
The replication traffic is often overlooked. With replication factor 3, each message is sent over the network twice (leader to follower 1, leader to follower 2). At 100MB/s ingest, that is 200MB/s of internal replication traffic on top of the 300MB/s consumer traffic. This is why messaging clusters need high-bandwidth networking. In cloud deployments, broker instances should be selected with network throughput as a primary criterion, not just CPU or memory.
Define the APIs expected from the system. This is your chance to analyze and define the read and write paths so that you can come up with the high-level design...
Three operations cover 90% of usage. But the parameters encode critical decisions: acks controls the durability-latency trade-off, key determines partition placement, and offset enables replay from any point in the log.
produce(topic, key, value, acks)
topic: The named channel to publish tokey: Determines which partition receives the message via hash(key) % num_partitions. Messages with the same key always land on the same partition, guaranteeing ordering for that key. If null, round-robin distribution is used.value: The message payload (bytes). The broker treats this as an opaque byte array. The broker does not parse or validate the content. Serialization format (JSON, Avro, Protobuf) is chosen by the application. Avro and Protobuf are preferred in production because they are compact (smaller messages = higher throughput) and support schema evolution.acks: Controls durability:acks=0: Fire and forget. Producer does not wait for any confirmation. Fastest, but messages can be silently lost.acks=1: Leader confirms. Producer waits for the partition leader to write to its local log. If the leader crashes before replication, the message is lost.acks=all: All in-sync replicas confirm. Producer waits for the leader AND all ISR followers to write. Safest, but adds 5-15ms latency.
fetch(topic, partition, offset, max_bytes)
topic, partition: Which partition to read fromoffset: The position in the log to start reading from. This is what makes the system replayable: consumers can seek to any historical position.max_bytes: Maximum data to return in one fetch. Combined with long polling (max_wait_ms), this lets consumers batch-read efficiently.The pull-based model means consumers call fetch repeatedly. Long polling (waiting up to max_wait_ms for new data) avoids busy-waiting while keeping latency low.
commitOffset(group_id, topic, partition, offset)
Consumers commit their progress after processing messages. On restart or rebalance, the consumer resumes from the last committed offset. Auto-commit (periodic background commits) is convenient but risks reprocessing if the consumer crashes between auto-commits. Manual commit gives precise control: commit only after the message is fully processed.
createTopic(name, partitions, replication_factor, retention_ms)
Creates a new topic with the specified configuration. The partition count and replication factor are set at creation time. Changing partitions later is possible but requires careful data migration since existing key-to-partition mappings change.
Describe the overall system architecture. Identify the main components needed to solve the problem end-to-end. Use the diagramming tool to create a block diagram.
The architecture has four major components: producers, the broker cluster, consumer groups, and the coordination service. Each plays a specific role, and a critical design decision is what is not in the architecture: there is no load balancer between producers and brokers.
Producers are client applications that publish messages. They maintain a cached copy of the partition map (which broker leads which partition) and send messages directly to the partition leader. When a producer first connects, it contacts any broker to fetch the full partition map. After that, it routes messages locally without additional network hops.
Brokers are the core of the system. Each broker accepts messages from producers, appends them to partition commit logs, replicates to follower brokers, and serves consumer fetch requests. A broker may be the leader for some partitions and a follower for others: roles are distributed across the cluster to balance load. A typical production deployment has 10-50 brokers. Each broker runs as a separate process on dedicated hardware (or a dedicated VM in cloud deployments). Brokers are stateless from a client perspective. all state is in the commit logs on disk and the coordination service. This means a failed broker can be replaced with a fresh instance that catches up by replicating from leaders.
Consumers organize into groups for parallel processing. Each partition is assigned to exactly one consumer in the group. If a consumer crashes, its partitions are reassigned to other group members: this is called a rebalance. Multiple consumer groups can read the same topic independently, each maintaining their own offsets. This is one of the most powerful features of the commit log model: adding a new consumer group (say, a search indexing service) requires zero changes to existing producers or consumers. The new group simply starts reading from the log at whatever offset it chooses.
An important constraint: you cannot have more active consumers in a group than partitions. If a topic has 12 partitions and you add a 13th consumer, that consumer sits idle: there is no partition to assign to it. This is why partition count planning (discussed in Capacity Estimation) matters.
The coordination service (ZooKeeper or KRaft) manages metadata: which brokers are alive, which broker leads each partition, where consumer offsets are stored, and what the topic configurations are. It also handles leader election when a broker fails. The coordination service does NOT handle message data. It only manages metadata. This separation is important: metadata is small (kilobytes) and changes infrequently, while message data is massive (terabytes) and changes continuously. Mixing them would create a bottleneck. The coordination service uses consensus (ZAB for ZooKeeper, Raft for KRaft) to ensure metadata is consistent across all nodes. even if a coordination node fails, the remaining nodes have an identical copy of all metadata.
Key Insight
Producers send messages directly to partition leaders, there is no load balancer or proxy in between. The partition map is cached client-side and refreshed from the coordination service when stale. This eliminates a middleman that would become a bottleneck at millions of messages per second. Every proxy hop adds latency and a failure point; direct-to-leader routing avoids both.
Level Expectations
Mid-level: Identify producers, brokers, consumers, and coordination service. Explain how partitioning enables parallel processing.
Senior: ISR replication with acks=all, consumer group rebalancing strategies (eager vs cooperative), pull-based backpressure.
Staff: Leader election protocol details, exactly-once via idempotent producers, cooperative rebalancing to avoid thundering herd, KRaft vs ZooKeeper trade-offs.
In a traditional web architecture, a load balancer distributes requests across servers. But a messaging system has a fundamental difference: messages must go to a specific partition leader, not any available server. A load balancer would need to inspect every message, compute the partition, and route accordingly, adding latency and becoming a single point of failure. Instead, producers do this routing themselves, using the locally cached partition map. When the map becomes stale (because a leader changed), the producer receives a NOT_LEADER_FOR_PARTITION error, refreshes its metadata, and retries. This self-healing pattern is simpler and faster than centralized routing.
Think about the scale: at 100K messages per second, a load balancer would need to inspect 100K messages, compute 100K partition hashes, and make 100K routing decisions. every second. Any single-node intermediary would become a throughput ceiling. By pushing routing logic to the producers (which are already distributed across many machines), the system scales linearly with the number of producers.
Define the data model. Identify the main entities, their attributes, and relationships. Consider the choice of database type (SQL vs NoSQL) and justify your decision based on access patterns...
Deep dive into 2-3 key components. Explain how they work, how they scale, discuss tradeoffs, capacity, and any relevant algorithms or data structures.
hash(key) % num_partitions determines the target partition. If no key, round-robin selects the next partition. The producer looks up the partition leader from its cached partition map.acks=all, the leader waits for all ISR followers to confirm their writes, then sends a success response to the producer. With acks=1, the leader responds immediately after its own write.If the partition leader has changed (broker failure, rebalance), the producer receives NOT_LEADER_FOR_PARTITION. The producer refreshes its partition map by contacting any broker, then retries the send to the new leader. This self-healing happens transparently, the application code does not need to handle it.
Producers accumulate messages in a buffer before sending. The linger.ms setting controls how long the producer waits to fill a batch (typically 5-50ms). Larger batches amortize more overhead but add latency. Compression (LZ4, Snappy, or zstd) is applied per batch, the entire batch is compressed as one unit, which achieves better compression ratios than compressing individual messages. The broker stores the compressed batch as-is; consumers decompress on their end. This reduces both network bandwidth and disk usage by 2-5x depending on message content.
The trade-off is clear: linger.ms=0 sends each message immediately (lowest latency, highest overhead), while linger.ms=50 batches messages for 50ms (higher latency, much better throughput). For most use cases, a small linger time (5-10ms) provides a good balance.
Consumer sends a fetch request with topic, partition, and offset. Broker reads from the commit log using zero-copy sendfile, returning messages directly from page cache to the network socket.
max_wait_ms parameter enables long polling, the broker holds the request until data is available or the timeout expires.sendfile() system call to transfer data directly from the page cache to the network socket, bypassing user-space memory copies. For recent data (still in page cache), this achieves near-memory speed. For older data, the OS reads from disk sequentially.__consumer_offsets internal topic.Common Pitfall
Zero-copy transfer via sendfile() only works when data is in the OS page cache. A lagging consumer reading data from days ago forces disk reads that consume I/O bandwidth, potentially degrading performance for all consumers, including those reading recent data. Monitor consumer lag closely and consider tiered storage for long-retention topics.
The core durability mechanism is In-Sync Replica (ISR) replication with acks=all. This is the single most important concept in the entire system. It is how we guarantee that no committed message is ever lost.
Producer sends to leader. Leader writes locally, replicates to ISR followers. Both followers acknowledge. Leader acks the producer. A follower that falls behind is removed from the ISR.
The ISR set is the leader plus all followers that are "caught up". their log end offset is within a configurable threshold of the leader's. With acks=all, the leader waits for ALL members of the ISR to write the message before acknowledging the producer.
What happens when a follower falls behind? The leader removes it from the ISR set. Subsequent acks=all writes only wait for the remaining ISR members. The fallen follower continues replicating in the background and is added back to the ISR once it catches up.
The min.insync.replicas setting adds a safety floor. With min.insync.replicas=2 and acks=all, the broker rejects writes if fewer than 2 replicas are in sync. This prevents the scenario where the ISR shrinks to just the leader (ISR size = 1), which would make acks=all equivalent to acks=1: a single point of failure.
The standard production configuration is replication factor 3 + min.insync.replicas 2 + acks=all. This tolerates exactly one broker failure while maintaining full durability guarantees. Two broker failures cause writes to be rejected (because the ISR drops below 2), which is the correct behavior: better to reject writes than to risk data loss.
Key Insight
GATE: acks=all combined with min.insync.replicas=2 guarantees that no committed message is ever lost. The leader only acknowledges after all ISR members write, and the ISR is never allowed to shrink below 2. Even if the leader crashes immediately after acknowledging, at least one follower has the message. This is a CP trade-off, we choose consistency (no data loss) over availability (reject writes when under-replicated).
Common Pitfall
Setting min.insync.replicas=1 with acks=all is false safety. If two followers lag and are removed from the ISR, the ISR shrinks to just the leader. Now acks=all means only the leader confirms, identical to acks=1. A single leader crash loses all messages written during this window. Always pair acks=all with min.insync.replicas=2.
Messages are assigned to partitions via hash(key) % num_partitions. Within a partition, messages are strictly ordered by offset: the broker appends messages sequentially and consumers read them in the same order.
But there is NO cross-partition ordering. If you send message A to partition 0 and message B to partition 1, consumers may see B before A. This is a fundamental trade-off: cross-partition ordering would require global coordination, destroying the parallelism that partitioning provides.
The practical implication: if ordering matters for a set of messages, they must share the same key (and therefore the same partition). For example, all events for a user session should use the session ID as the key.
What about ordering across an entire topic? Some systems need total ordering. every consumer sees every message in the same global order. The only way to achieve this is to use a single partition. But a single partition means a single consumer, a single leader broker, and no parallelism. At 10-50MB/s per partition, this caps throughput severely. In practice, total ordering is rarely needed: per-key ordering (all events for user X in order) is almost always sufficient, and per-key ordering is free with partitioning.
Three consumers handle six partitions. C3 crashes. The coordinator detects the missed heartbeat and triggers a cooperative rebalance, reassigning only C3 partitions to C1 and C2.
When a consumer joins or leaves a group, partitions must be redistributed. Two strategies exist:
Eager rebalancing (stop-the-world): ALL consumers release ALL partitions. The coordinator reassigns everything from scratch. During this window, NO messages are consumed. This is simple but causes a consumption pause proportional to the number of partitions.
Cooperative rebalancing (incremental): Only the affected partitions are revoked and reassigned. Unaffected consumers continue processing without interruption. This minimizes the blast radius: if one consumer crashes, only its partitions experience a pause.
Cooperative rebalancing prevents the thundering herd problem: with eager rebalancing, a single consumer crash causes ALL consumers to stop, which may trigger heartbeat timeouts for healthy consumers, causing further crashes and cascading rebalances. In a large consumer group with hundreds of partitions, this cascading failure can take minutes to stabilize: during which zero messages are processed. Cooperative rebalancing eliminates this entirely by limiting the blast radius to only the affected partitions.
Producer sends a message with producer ID and sequence number 5. Network times out. Producer retries the same message with sequence 5. Broker detects the duplicate and returns success without writing twice.
Network timeouts create a dangerous ambiguity: did the broker receive the message or not? If the producer retries, a naive broker would write the message twice. Idempotent producers solve this.
Each producer is assigned a unique producer ID on initialization. For each partition it sends to, the producer maintains a monotonically increasing sequence number. The broker tracks the latest sequence number per producer-partition pair. If a message arrives with a sequence number that has already been committed, the broker rejects the duplicate and returns success (so the producer does not retry again).
This makes individual produce operations exactly-once: even if the producer retries due to timeouts, the broker guarantees no duplicate writes. Combined with transactional commits, this extends to atomic multi-partition writes.
Consumer offsets can be stored in two places: the coordination service (ZooKeeper) or an internal Kafka topic (__consumer_offsets). Modern deployments use the internal topic, which is itself a compacted log. only the latest offset per consumer-group-partition key is retained.
When a consumer crashes or a rebalance occurs, the new consumer reads the last committed offset from __consumer_offsets and resumes from that point. Messages between the last commit and the crash are reprocessed (at-least-once semantics). To minimize the reprocessing window, consumers can commit more frequently: but more frequent commits add overhead to the coordination service.
The practical recommendation: commit after processing each batch of messages. This limits the reprocessing window to one batch size (typically a few hundred messages) while keeping commit overhead manageable. For exactly-once semantics, use transactional consumers that atomically commit offsets and processing results.
The acks parameter is the most important trade-off in the entire system. Here are concrete latency numbers for a typical production cluster:
The difference between acks=1 and acks=all is only 3-10ms: a small price for guaranteed durability. In most production systems, acks=all with min.insync.replicas=2 is the correct default.
Pull-based (Kafka model): Consumers request messages when ready. Natural backpressure, a slow consumer simply polls less frequently. The downside is polling overhead, but long polling (max_wait_ms) eliminates this by holding the request until data arrives. Pull-based is better for high-throughput scenarios where consumers process at varying rates.
Push-based (RabbitMQ model): The broker sends messages to consumers as they arrive. Lower latency for light loads because there is no polling delay. The downside is that the broker must track delivery state per consumer, and a slow consumer can be overwhelmed. Push-based is better for task distribution where each message is a unit of work consumed exactly once.
Most distributed messaging systems choose pull-based because the throughput advantages outweigh the latency difference, especially when long polling is used.
ZooKeeper is a separate distributed coordination service. It stores cluster metadata reliably using ZAB consensus. The drawback is operational complexity: you must deploy, monitor, and maintain a 3-5 node ZooKeeper ensemble alongside the Kafka cluster. ZooKeeper also becomes a scalability bottleneck because it stores all partition metadata in memory, limiting clusters to roughly 200,000 partitions.
KRaft (Kafka Raft) eliminates ZooKeeper entirely. A subset of Kafka brokers are designated as controllers. They form a Raft quorum to manage metadata. Metadata is stored in an internal Kafka topic, replicated like any other topic. KRaft supports millions of partitions (10x ZooKeeper's limit), simplifies deployment (one system instead of two), and reduces operational overhead. The trade-off is maturity. KRaft reached production readiness in 2023, so some older deployments still use ZooKeeper.
More partitions enable more consumer parallelism. But the costs grow:
The sweet spot is typically 10-50 partitions per high-traffic topic. Start with fewer partitions and increase when consumer parallelism becomes the bottleneck. You can always add partitions, but you cannot remove them without recreating the topic.
Two retention strategies exist for old data:
Time/size-based deletion (default): Entire log segments older than the retention period (or exceeding the size limit) are deleted. Simple and predictable. just delete old files. Best for event streams where historical data has a natural expiry (7-day clickstream, 30-day audit logs).
Log compaction: Instead of deleting old segments, the broker keeps only the latest value for each key. Older values are removed during background compaction. The result is a snapshot of the latest state per key. Best for changelog topics (database CDC, configuration updates) where consumers need the current state, not the full history. Compacted topics never lose the latest value for any key, regardless of age.