Bird
Raised Fist0

Consistent Hashing for Database Sharding at Scale

HLDmediumInterview freq: common
Choose your preparation mode4 modes available
ArchitectureTraceChallenge
🎯
Consistent Hashing for Database Sharding at Scale
45 min · senior
senior

You are building a scalable database sharding layer for a startup with 50 million daily active users (DAU). The system must distribute user data evenly across hundreds of database nodes while supporting dynamic node addition/removal without significant data reshuffling. The business requires encrypted data at rest and in transit, offline data delivery for mobile clients, and a P99 latency under 200ms for read/write operations.

Constraints
  • Data must be encrypted at rest and in transit using TLS and AES-256 encryption
  • Support dynamic scaling: nodes can be added or removed without full data rebalancing
  • Offline data delivery for mobile clients with eventual consistency guarantees
  • P99 latency for read/write operations must be under 200ms
  • System must handle 50M DAU with expected peak QPS of 1M writes and 5M reads
  • Data distribution must minimize hotspots and ensure even load across shards
📋
Guided Steps
1/5
Step 1
Choosing the Sharding Strategy
What sharding strategy will you use to distribute data evenly and support dynamic node membership?
💡 Hint
Consider a hashing mechanism that minimizes data movement when nodes change.
Key Decision
Use consistent hashing with virtual nodes implemented via a hash ring (e.g., Ketama algorithm) to evenly distribute keys and minimize remapping on node changes.
Interviewer Expectation
Mid: Names consistent hashing. Senior: Explains virtual nodes to reduce load imbalance and how consistent hashing reduces data reshuffling compared to modulo-based sharding.
⚠️Common Wrong Answer
Use modulo-based sharding (key % N) - causes massive data reshuffling and hotspots when nodes are added or removed.
Step 2
Mapping Keys to Database Nodes
How will you map user keys to physical database nodes and handle node failures or additions?
💡 Hint
Think about virtual nodes and a distributed metadata service for node membership.
Key Decision
Implement a consistent hash ring with multiple virtual nodes per physical node stored in ZooKeeper or etcd for membership coordination and failure detection.
Interviewer Expectation
Mid: Mentions consistent hashing with virtual nodes. Senior: Details how ZooKeeper/etcd maintain node membership, detect failures, and trigger client metadata refresh to avoid stale routing.
⚠️Common Wrong Answer
Hardcode node list in clients - leads to stale routing info and inconsistent reads/writes during node changes.
Step 3
Ensuring Low Latency and Encryption
How will you ensure P99 latency under 200ms while encrypting data at rest and in transit?
💡 Hint
Consider TLS for transport and native database encryption features; also think about caching and connection pooling.
Key Decision
Use TLS 1.3 for all client-server and inter-node communication, enable AES-256 encryption at rest via database native encryption (e.g., AWS KMS with Cassandra Transparent Data Encryption), and deploy Redis as a local cache with connection pooling to reduce read latency.
Interviewer Expectation
Mid: Mentions TLS and encryption at rest. Senior: Explains TLS 1.3 benefits, native DB encryption overhead, and how Redis caching reduces read latency to meet P99 SLA.
⚠️Common Wrong Answer
Encrypt only in transit or use application-level encryption without hardware acceleration - causes latency spikes and CPU overhead.
Step 4
Handling Offline Data Delivery and Eventual Consistency
How will you support offline data delivery for mobile clients with eventual consistency?
💡 Hint
Think about write-ahead logs, change data capture, and client sync protocols.
Key Decision
Use Apache Kafka as a change data capture (CDC) pipeline to stream database mutations; mobile clients sync via a cursor-based API that reads from Kafka-backed storage (e.g., Cassandra) to deliver offline updates with eventual consistency.
Interviewer Expectation
Mid: Names Kafka for CDC. Senior: Explains cursor-based sync, how Kafka partitions align with shards, and how clients reconcile conflicts using vector clocks or timestamps.
⚠️Common Wrong Answer
Use polling without CDC - causes excessive load and inconsistent offline data.
Step 5
API Design for Shard Routing and Membership Updates
Design 3 core API endpoints for clients to write data, read data, and handle shard membership changes.
💡 Hint
Consider endpoints for write(key, data), read(key), and getShardMetadata() or subscribeShardChanges().
Key Decision
Expose Write(key, data) and Read(key) endpoints that internally hash keys to shards; provide a GetShardMetadata() endpoint returning current hash ring and node info, plus a subscription mechanism (e.g., WebSocket or SSE) for clients to receive membership updates.
Interviewer Expectation
Mid: Defines basic read/write APIs. Senior: Details consistent hashing integration, metadata caching, and real-time membership update delivery to avoid stale routing.
⚠️Common Wrong Answer
No membership update API - clients use stale shard info causing routing errors and increased latency.
🔌
API Design Challenge
Sketch 3 core API endpoints for the sharding layer that clients and internal services will use.
Hints
  • How does a client write data and specify the shard key?
  • How does a client read data ensuring consistent routing to the correct shard?
  • How does the system handle shard membership changes and notify clients?
📊
Scoring Rubric
▼ Show
Must Have - 60%
  • Consistent hashing with virtual nodes to minimize data movement
  • Distributed membership coordination using ZooKeeper or etcd
  • Encryption in transit (TLS 1.3) and at rest (AES-256 with native DB support)
  • API design including read, write, and shard metadata endpoints
Good to Have - 30%
  • Use of Redis caching to meet latency SLAs
  • Kafka-based CDC pipeline for offline data delivery
  • Cursor-based incremental sync for mobile clients
  • Subscription mechanism for shard membership updates
Bonus - 10%
  • Detailed failure handling for node removal and data rebalancing
  • Explanation of virtual node count tuning to balance load
  • Discussion of conflict resolution strategies for eventual consistency
Anti-patterns to Avoid
Modulo-based sharding causing massive data reshuffling and hotspots on node changes, leading to high latency and downtime during scaling events
Hardcoding shard membership in clients leading to stale routing and inconsistent data access, causing read/write errors and increased latency
Encrypting only in transit or using inefficient application-level encryption causing latency spikes and CPU overhead, degrading system performance
Polling-based offline sync causing excessive load and poor user experience due to high resource consumption and delayed updates
No shard membership update API causing routing errors and increased latency as clients operate on stale shard information
Ignoring load balancing leading to hotspots and degraded system throughput, resulting in uneven resource utilization and potential outages
✅
Model Answer
▼ Reveal

This design uses consistent hashing with virtual nodes (Ketama) to distribute user data evenly across hundreds of database shards. ZooKeeper or etcd coordinate shard membership and detect failures, enabling dynamic scaling without full data reshuffling. TLS 1.3 secures data in transit, and AES-256 encryption at rest is enabled via native database features like Cassandra Transparent Data Encryption. Redis caching with connection pooling reduces read latency to meet the P99 SLA. Apache Kafka streams database changes for offline data delivery, with mobile clients syncing incrementally via a cursor-based API. The sharding layer exposes Write(key, data), Read(key), and GetShardMetadata() APIs, with a subscription mechanism for real-time shard membership updates to avoid stale routing.

Key Decisions
  • Use Ketama consistent hashing with 100+ virtual nodes per physical node to balance load and minimize remapping on node changes
  • Store shard membership and virtual node mappings in ZooKeeper/etcd to enable dynamic membership updates and failure detection
  • Encrypt all network traffic using TLS 1.3 to reduce handshake latency and improve security
  • Enable AES-256 encryption at rest using Cassandra Transparent Data Encryption integrated with AWS KMS for key management
  • Deploy Redis as a local cache with connection pooling to reduce read latency and meet P99 SLA under high QPS
  • Implement Apache Kafka CDC pipeline to stream database mutations for offline data delivery and eventual consistency
  • Provide a cursor-based sync API for mobile clients to fetch incremental updates efficiently
  • Expose Write(key, data) and Read(key) APIs that hash keys to shards internally, plus GetShardMetadata() and a subscription endpoint for real-time shard membership updates
  • Consistent hashing with virtual nodes (Ketama) to minimize data reshuffling on node changes
  • ZooKeeper or etcd for distributed shard membership coordination and failure detection
  • TLS 1.3 for encrypted transport and AES-256 encryption at rest using native DB features (e.g., Cassandra TDE)
  • Redis as a local cache with connection pooling to reduce read latency and meet P99 SLA
  • Apache Kafka for change data capture enabling offline data delivery and eventual consistency
  • Cursor-based sync API for mobile clients to efficiently fetch incremental updates
  • Shard metadata service exposing hash ring info and membership changes via subscription APIs
  • Use of virtual nodes to balance load evenly across hundreds of physical database nodes

Practice

(1/5)
1. In the webhook delivery system, at Step 4 the Delivery Worker checks the subscriber's last acknowledged event ID before sending the next event. What state does the Delivery Worker hold if the subscriber's acknowledgement is delayed beyond the retry timeout?
easy
A. Immediately escalates the failure to the alerting system
B. Keeps the event in 'pending retry' state and schedules a retry
C. Deletes the event from the queue assuming subscriber is unreachable
D. Marks the event as delivered and moves on to the next event

Solution

  1. Step 1: Trace Delivery Worker decision at Step 4 -- subscriber ack delay

    The worker waits for acknowledgement; if delayed beyond retry timeout, it escalates failure to alerting for manual intervention.

  2. Step 2: Eliminate other options

    Marking as delivered without ack breaks reliability; deleting event loses data; scheduling retry is normal but after multiple failures escalation is needed.

  3. Final Answer:

    Option A -> Option A

  4. Quick Check:

    Delayed ack beyond retry timeout -> escalate alert [check]
Hint: Delayed ack triggers alert escalation [check]
Common Mistakes:
2. A web application experiences increasing CPU load on a single database server, causing query latency spikes. Which scaling approach best addresses this bottleneck without changing the application code?
easy
A. Distribute the database load by sharding data across multiple servers (horizontal scaling)
B. Add more CPU cores and RAM to the existing database server (vertical scaling)
C. Use a caching layer in front of the database to reduce read load
D. Switch from SQL to NoSQL database to improve write throughput

Solution

  1. Step 1: Identify the bottleneck as CPU and memory on a single DB server

    Vertical scaling adds resources to the existing server to handle more load without architectural changes.

  2. Step 2: Evaluate horizontal scaling and other options

    Horizontal scaling requires application changes for sharding; caching reduces reads but not CPU-bound writes; switching DB types is a different trade-off.

  3. Final Answer:

    Option B -> Option B

  4. Quick Check:

    Vertical scaling adds resources to one server without code changes [check]
Hint: Vertical scaling adds power to one machine [check]
Common Mistakes:
3. If the Road Graph Service's update pipeline fails to process live traffic changes, what is the exact failure observed in ETA computation?
medium
A. ETA computations continue using outdated traffic data, causing inaccurate arrival times
B. Users receive error messages indicating routing data is unavailable
C. Routing Engine crashes due to missing graph updates, causing system downtime
D. The system automatically switches to a backup graph with no traffic data, causing longer routes

Solution

  1. Step 1: Understand Road Graph update pipeline failure impact

    Without updates, live traffic data is stale; system must handle this gracefully.

  2. Step 2: Trace failure effect on ETA computation

    Routing Engine does not silently use stale data (A) or crash (B); it returns errors (C) to avoid misleading ETAs; no automatic backup switch (D).

  3. Final Answer:

    Option B -> Option B

  4. Quick Check:

    Update pipeline failure causes routing errors, not silent stale data use [check]
Hint: Update failure triggers routing errors, not silent stale data use [check]
Common Mistakes:
4. At 10x transaction volume, which component in the digital wallet system is most likely to become the first bottleneck, and why?
hard
A. The event log (Kafka), because partition count limits throughput and rebalancing partitions causes downtime
B. The reconciliation service, because it performs expensive cross-checks on all transactions synchronously
C. The API gateway, because it handles all client requests and cannot horizontally scale due to session affinity
D. The ledger database, because it requires strong consistency and cannot shard entries without breaking double-entry guarantees

Solution

  1. Step 1: Analyze ledger database constraints

    Ledger DB enforces strong consistency and atomic double-entry transactions, limiting sharding options.

  2. Step 2: Compare other components

    Reconciliation (A) can be scaled asynchronously; API gateway (C) can be scaled with stateless design; Kafka (D) scales with partitioning but ledger DB consistency is harder to scale.

  3. Final Answer:

    Option D -> Option D

  4. Quick Check:

    Ledger DB consistency limits sharding -> first bottleneck [check]
Hint: Ledger DB consistency limits sharding -> first bottleneck [OK]
Common Mistakes:
5. At 10x scale of global traffic, which component in the BGP-based routing infrastructure is most likely to fail first, and why?
hard
A. Edge routers, because their CPU cannot handle the increased packet forwarding rate
B. Route reflectors, because they become overwhelmed maintaining full routing tables and propagating updates
C. Network monitoring systems, because data volume exceeds processing capacity causing delayed alerts
D. Client devices, because they cannot process rapid route changes at scale

Solution

  1. Step 1: Identify bottleneck at scale

    Edge routers handle packet forwarding at line rate; at 10x traffic, their CPU and forwarding ASICs are stressed first.

  2. Step 2: Evaluate other components

    Route reflectors (A) scale better with hierarchical design; monitoring systems (C) can be scaled horizontally; client devices (D) are downstream and less critical for routing stability.

  3. Final Answer:

    Option A -> Option A

  4. Quick Check:

    Edge routers' CPU overload causes first failure at 10x traffic [check]
Hint: Edge routers hit CPU limits first at massive scale [check]
Common Mistakes:
Recommended

Ready to prove it?

Pass the test and this pattern moves unknown strong while earning +1 Grip toward your certificate.

10 questions~5 mineasy hard