Rebalancing Implementation: Algorithms, Execution, and Observability

Implementing production grade rebalancing requires careful coordination of placement algorithms, execution plans, and observability. Consistent hashing with virtual nodes typically assigns 64 to 256 tokens per physical node, weighted proportionally to capacity. For a 100 node cluster with 128 tokens per node storing 1 PB total (10 TB per node average), adding 10 equal capacity nodes triggers redistribution of approximately 100 TB total. The algorithm identifies keys assigned to tokens now owned by new nodes and marks them for movement. With 100 TB to move at a throttled 250 MB per second effective throughput per active move and 2 concurrent moves per node, expect multi hour completion: moving 1 TB takes roughly 1.1 hours at 250 MB per second, so 100 TB with parallelism across the fleet takes several hours. Execution plans for stateful shards involve multiple phases to maintain availability. First, pre copy the bulk of data while the source shard continues serving traffic, streaming snapshots to the target. Second, enter a catch up phase where changelog or replication streams delta updates until lag falls below a threshold (typically under 1 second or 10 MB). Third, perform cutover: for leader moves, fence the old primary, promote the replica, and update routing; for replica moves, simply update replica set membership. Finally, clean up the old shard after confirming the new one is stable. Rate limiting is critical: cap background copy bandwidth to 50 to 100 MB per second per disk to avoid starving foreground reads and writes. Amazon and LinkedIn systems instrument every phase with timeouts and rollback conditions: if catch up phase exceeds 10 minutes, abort and retry later. Lease based consumer systems use heartbeat intervals and session timeouts to balance failure detection speed against false positives. A common configuration is 5 to 10 second heartbeats with 15 to 30 second session timeouts, meaning a failed worker's leases expire and become available for reassignment within 15 to 45 seconds. The tradeoff is detection latency versus stability: shorter timeouts enable faster recovery from real failures but risk spurious rebalances from transient network delays or garbage collection pauses. Staggering heartbeat polls across consumers avoids thundering herds where 100 consumers all query the coordination store simultaneously, overwhelming it. Observability separates good implementations from brittle ones. Key metrics include moved bytes, moves inflight (current concurrent operations), time to balance (how long from trigger to completion), fraction of assignments changed (for cooperative rebalancing), per node utilization (CPU, disk, network), consumer lag (for stream systems), and latency percentiles (p95, p99) during rebalances. Emit reason codes for every rebalance trigger: node add, node remove, hot shard detected, Availability Zone constraint violation. At Amazon scale, dashboards track these metrics across thousands of clusters; anomaly detection alerts when time to balance exceeds historical norms or when rebalance frequency spikes, indicating potential storms before they cascade.

100 node cluster with 128 virtual tokens per node storing 1 PB: adding 10 nodes moves approximately 100 TB total (10 TB per new node), at 250 MB per second with 2 concurrent moves per node, rebalancing completes in 4 to 6 hours

Kinesis worker with 10 second heartbeat and 30 second session timeout: worker crashes, leases expire after 30 seconds (3 heartbeat intervals), other workers detect and claim shards within 30 to 45 seconds, consumer lag increases briefly

Elasticsearch shard relocation dashboard: shows 8 moves inflight (4 nodes, 2 concurrent each), average bandwidth 75 MB per second per move, time to balance estimate 2.5 hours for remaining 40 shards, p99 search latency stable at 25 milliseconds