System requirements

Functional Requirements:

• Asynchronous communication between services.
• Support multiple messaging patterns:
  • Publish/Subscribe (Pub/Sub) – multiple subscribers can receive messages from a topic.
  • Point-to-Point / Queue – messages delivered to a single consumer.
  • Request/Reply – synchronous response pattern.
• Reliable message delivery:
  • At-least-once, at-most-once, exactly-once delivery semantics.
• Message ordering guarantees per topic or partition.
• Message persistence and recovery.

Non-Functional Requirements:

• Scalability: handle millions of messages/sec.
• Low Latency: minimal delay between producer and consumer.
• Fault Tolerance: survive node failures.
• Monitoring and Observability: track message flow, failures, and performance metrics.

API design

Offer both gRPC and HTTP/REST endpoints for admin and simple produce; clients primarily use native language clients (high-performance TCP binary protocol over TLS).

High-level design

Producers — publish messages to topics/queues.

- Connects to broker leader for target partition

- Supports batching, retries, idempotence

- Assigns keys for partitioning and ordering

- Uses acknowledgments to confirm durable write

Brokers — cluster of servers responsible for storing and serving messages (log-based storage).

- Maintains commit logs (append-only segments)

- Handles replication to follower brokers

- Routes messages to consumers

- Tracks offsets and consumer group states

- Performs compaction and retention cleanup

Metadata/Coordination service — leader election, cluster membership, topic metadata (can be consensus-based: Raft/ZooKeeper/etcd).

- Leader election for topic partitions

- Tracks broker liveness (heartbeats)

- Manages consumer group membership

- Stores configuration (topic metadata, offsets)

- Detects and triggers failover

Consumers — pull or receive messages; support consumer groups (competing consumers).

- Subscribes to topics (via group)

- Fetches from assigned partitions

- Acknowledges or commits offsets

- Handles redelivery and recovery

- Supports different ack modes (auto/manual)

Coordinator/Controller — manages partitions, rebalances consumers, enforces quotas, controller leader.

Storage layer — append-only log files per partition with segmenting, compaction, retention policies.

- Stores messages in partitioned commit logs on disk

- Handles log segments, indexing, compaction

- Replicates data across brokers for durability

Replicator — intra-cluster replication (sync/async) for high availability.

Delivery & offset manager — track per-consumer offsets/ACKs.

Admin API & Management UI — create topics, set retention, view status.

Observability stack — metrics, tracing, logging, alerting.

Clients SDKs — for major languages, supporting batching, compression, retries, transactions.

Storage model:

• Append-only log per partition (log-structured). Messages are appended; consumers read at offsets.
• Each topic split into partitions to scale throughput, each partition has a leader and replicas.

Consensus/replication:

• Use Raft or Kafka-like ISR with leader election to replicate partition log entries. Replication factor ≥3 recommended.
• Controller uses Raft to manage metadata and coordinate reassignments.

Fault tolerance:

• Leader failover: new leader chosen from in-sync replicas.
• Producer acks: configurable (leaderAck, allReplicasAck).
• Consumer offset commit stored in compacted topic (or external store) to enable recovery.

Sequence (produce/consume):

• Producer -> broker leader -> append, replicate -> ack -> consumer polls leader -> process -> commit offset.

Detailed component design

Producers

Responsibilities:

• Serialize messages (JSON, Avro, Protobuf).
• Partitioning logic: decide which partition a message goes to (hashing by key or round-robin).
• Retry and batching:
  • Buffer messages for batch writes to reduce I/O.
  • Retry failed messages until acknowledged (if at-least-once delivery is desired).
• Optional compression to reduce network/broker load.

Brokers

Brokers are the heart of the system. Each broker stores partitions, handles replication, and serves consumers.

Responsibilities:

• Accept messages from producers.
• Append messages to partition logs.
• Replicate partition logs to follower brokers.
• Serve consumer requests (read messages from offset).
• Maintain metadata (leader/follower role for partitions).
• Garbage collection: clean up expired messages (retention policy).
• Provide fault tolerance: detect leader failure and promote followers.

Consumers

Consumers read messages from brokers and acknowledge progress.

Responsibilities:

• Fetch messages from brokers for assigned partitions.
• Commit offsets to mark processed messages.
• Support consumer groups for load balancing:
  • Each partition is consumed by only one consumer in a group.
  • Rebalancing when consumers join/leave.
• Handle retries/failures and optional idempotency.

Coordination / Metadata Service

Purpose: Maintain cluster state and enable consistent partition leadership.

Responsibilities:

• Track cluster nodes (brokers, consumers).
• Manage leader election for partitions.
• Store topic configurations and partition mappings.
• Track consumer offsets (if centralized).

Typical Implementations:

• ZooKeeper, etcd, Consul.
• In modern Kafka versions, KRaft mode replaces ZooKeeper with internal metadata quorum.

Data Stored:

• List of brokers and their status.
• Partition → broker mapping (leader + replicas).
• Topic metadata (name, replication factor, retention).
• Consumer group membership and last committed offsets.

Storage Layer

Purpose: Efficiently persist messages with durability and fast access.

Design:

• Partitioned log segments on disk:
  • Each partition is split into multiple log segments (e.g., 1GB files).
• Index files:
  • Offset index: maps offset → byte position in log file.
  • Timestamp index: maps timestamp → offset.
• Replication:
  • Each partition has a leader broker and multiple follower replicas.
  • Followers replicate leader logs asynchronously.
  • Quorum-based acknowledgment for durability.

Monitoring & Observability

Responsibilities:

• Collect metrics: throughput, latency, consumer lag.
• Alert on failures: broker down, replication lag, consumer stalled.
• Logging: message failures, retries, and offsets.
• Tools: Prometheus, Grafana, ELK stack.

Messaging patterns supported

1. Publish/Subscribe (topic, fan-out)
   • Many producers publish to a topic; many subscribers (consumer groups) get messages. Consumers in the same consumer group share partitions (competing consumers). Multiple consumer groups get full copy (fan-out).
2. Work queues
   • Use a topic with N partitions and a consumer group with M consumers. Each message delivered to one consumer in group (load balancing).
3. Request/Reply
   • Client produces request message with reply_to topic + correlation_id. Server consumes, processes, produces response to reply_to topic with same correlation_id. Client consumes responses filtered by correlation id (or uses temporary exclusive reply queue). Provide timeouts and idempotence on server.
4. Exactly-once processing
   • Use producer idempotence, transactional produce+consume (consume->process->produce within a transaction), and broker transactional support to atomically commit consumer offsets and produced messages. Works best when both cluster and clients support transactions.
5. Delayed/TTL & Dead-letter
   • Support TTL per message or per topic. Failed messages after N retries go to DLQ with metadata about failure.

Delivery semantics & guarantees

• At-most-once: producer sends, no retries; consumer auto-ack before processing. Rarely used for critical data.
• At-least-once (default): broker persists until ack; consumers may reprocess after crash → idempotent consumers recommended.
• Exactly-once (end-to-end): requires idempotent producers + broker transactions + atomic offset commits. Stronger overhead; recommend for financial events.

Config knobs:

• acks (0/1/all), retries, enable.idempotence, transactional.id, min.insync.replicas, replication.factor, retention.ms, cleanup.policy.

Partitioning, scaling & performance

• Partitioning: horizontal scale via partitions. Higher partitions → more throughput but higher coordination overhead.
• Partition assignment: hash(key) % num_partitions. For ordered processing per key, ensure same key maps to same partition.
• Throughput: tune batch.size, linger.ms, compression (lz4/snappy), acks=all for durability.
• Latency vs throughput trade-offs: larger batches and compression increase throughput but add latency.
• Scaling brokers: add brokers and reassign partitions evenly. Use controller to rebalance.

Capacity planning formula (simplified):

• required_write_throughput_bytes_per_sec = avg_msg_size * p50_msgs_per_sec
• partitions_needed ≈ ceil(required_throughput_per_topic / per_partition_max_throughput)
• per_partition_max_throughput measured under typical hardware and IO (test).

Replication & ISR:

• Replication factor 3, min.insync.replicas=2 ensures writes succeed only if at least 2 replicas in sync.

Hardware:

• Fast NVMe SSDs (append-only log), enough RAM for cache, NIC 10Gb+ for high throughput.

Message tracking, tracing, replay & recovery

• Trace IDs: require trace_id header for distributed tracing. Integrate with Jaeger/Zipkin.
• Schema Registry: store schemas and compatibility rules. Consumers validate incoming messages.
• Message index / metadata DB: compacted topics for metadata (e.g., keys → last offset) if needed.
• Replay: consumers can reset offsets to a timestamp or offset to reprocess.
• Retention / snapshots: optional long-term retention (S3 as tiered storage). Use cold storage for audit and replay (e.g., segment uploads).
• DLQ: store failing messages with failure reason and attempt count.

Recovery flows:

• Consumer failure: restart consumer; it reads last committed offset and resumes.
• Broker failure: partition leader fails → controller elects new leader.
• Full cluster failover: use cross-dc replication (async) to promote mirror cluster.

Auditing:

• Keep an audit log topic with message metadata (timestamp, producer id, offset, topic, schema_id) to reconstruct flows.

Database design

Database primarily needs to store:

• Messages.
• Topics and Partitions.
• Consumers and Consumer Groups.
• Offsets / Delivery tracking.
• Broker / Cluster metadata.

Schema:

Messages (the actual payloads) - store in append-only log (custom file storage) or distributed commit log (Cassandra).

Metadata - store in relational DB (PostgreSQL, MySQL) or Key-Value store (etcd, Consul, ZooKeeper).

Offsets / Consumer Tracking - store in Key-Value store (Redis).

Monitoring & Logs - Time-series DB (Prometheus, InfluxDB).