Rebalance Storms and the Backpressure That Prevents Them

A rebalancer that responds to every imbalance immediately is unstable. Each move consumes the bandwidth and CPU that the cluster needs to serve requests; the resulting latency causes clients to retry; the retries raise load, which makes the imbalance look worse; and the rebalancer responds by moving more data. The loop has no fixed point until something fails. Preventing it is a control problem, and the control has to be built into the rebalancer rather than bolted on after the first incident.

The failure that starts it

A single node holding P partitions fails. Every one of those partitions is now under-replicated or entirely unavailable, and the recovery plan is to re-create each one on a surviving node and copy its data from a replica or from the failed node’s disk.

The data volume is the failed node’s share of the physical copy count. With replication factor R spread over N nodes, each node holds about R/N of the logical data, so restoring replication means writing one new copy of that share. For R = 3 and N = 100, that is about 3 per cent of the logical volume: a 100 TB data set needs roughly 3 TB of transfer to restore, and the cluster’s aggregate spare bandwidth is what bounds how long that takes. The surviving replicas are not re-copied; only the missing one is.

The recovery is also concurrent. All P partitions want to copy at once, and each copy has a source node that is still serving requests. The sources are a subset of the surviving nodes, often the ones that held the other replicas, and they are exactly the nodes that are now absorbing the read and write traffic of the failed node’s share. They are the busiest nodes in the cluster, and they are being asked to stream data out at maximum rate. The protocol each of those copies runs is the migration state machine described in Shard Migration Cutover and the Dual-Write Window, and it is not designed to be run hundreds of times at once.

This is the structural reason recovery is dangerous: the nodes that must supply the recovery data are the nodes that are already overloaded by the failure.

The amplification loop

The loop has four links, and each is individually reasonable.

Migration traffic competes with request traffic on the same network link. A node streaming a partition at 800 Mbit/s on a 1 Gbit/s link leaves 200 Mbit/s for requests, which increases request latency.

Higher latency causes client timeouts. A client with a one-second timeout and a server whose p99 rises above one second will time out a fraction of its requests.

Timeouts cause retries, and retries are additional requests. A retry is a full request, so a timeout rate of 5 per cent adds 5 per cent to the request load, which raises latency further, which raises the timeout rate. The feedback factor is the product of the timeout rate and the retry count; if each timed-out request is retried three times, a 5 per cent timeout rate adds 15 per cent load, which is enough to push a node at 85 per cent utilization past saturation.

Saturation makes the rebalancer move more data. A node that is overloaded looks imbalanced, because its queue is long and its response time is high. A rebalancer that uses latency as its imbalance signal will decide that the node holds too many partitions and start moving them off, which adds more migration traffic to the node it is trying to relieve.

The fourth link is what makes the loop a storm rather than a transient. The first three links are self-limiting if the rebalancer is driven by data volume or partition count, because those metrics do not change with load. They are not self-limiting if the rebalancer is driven by latency or by request rate, because those metrics worsen as the loop turns.

Why the loop is not fixed by a bigger timeout

Lengthening the client timeout breaks one link at the cost of another. A timeout of ten seconds tolerates the latency spike, so retries do not amplify. But a ten-second timeout also means a request that would have failed fast now occupies a connection and a server thread for ten seconds, which increases the concurrency on the server and worsens the queueing. The timeout is a load-shedding parameter, not just a patience parameter, and raising it converts a retry storm into a queue-depth problem.

The correct response is to shed load explicitly rather than to let timeouts do it implicitly. A server that is over its concurrency limit should reject requests immediately with a retryable error and a hint, rather than queueing them until they time out. An immediate rejection costs the client one round trip and costs the server nothing, whereas a queued request costs the server memory and a slot in the queue, and delays every request behind it.

Admission control for migration

The migration traffic must be admitted through the same control as request traffic, and it must lose to request traffic when the node is saturated.

The mechanism is a token bucket per node, applied to migration streams, with the rate set as a fraction of the link’s capacity and adjusted by the node’s request latency. When the node’s request latency exceeds a target, the migration rate is reduced; when latency is below target, it is increased. This is a closed-loop controller, and its stability depends on the loop’s delay being short relative to the timescale of the load changes. The latency being controlled is itself the fan-out latency of requests that touch many partitions, which is why a target set from single-partition measurements is far too tight for a coordinator that gathers across the cluster, as Scatter-Gather Fan-Out and Tail Latency Multiplication describes.

class MigrationThrottle:
    def __init__(self, link_bps, target_ms, min_share=0.05, max_share=0.5):
        self.link = link_bps
        self.target = target_ms
        self.share = 0.2                 # start at 20 per cent of the link
        self.min, self.max = min_share, max_share

    def observe(self, p99_ms):
        # multiplicative decrease, additive increase, in log space
        if p99_ms > self.target * 1.2:
            self.share = max(self.min, self.share * 0.7)
        elif p99_ms < self.target * 0.8:
            self.share = min(self.max, self.share * 1.05)
        return int(self.link * self.share)

The asymmetry in the update is deliberate. A large cut when latency is high, a small increase when it is low, so that the controller converges toward the highest rate that does not disturb the node. The minimum share guarantees that migration never stops entirely, because a migration that stalls indefinitely leaves the cluster under-replicated, which is a worse failure than slow recovery. The maximum share caps migration at half the link so that requests always have capacity.

The controller’s target must be the node’s own request latency, not the cluster’s, because the node’s capacity is what the controller is trying to protect. A cluster-wide average would hide the one node that is saturated, which is the node whose migration stream needs throttling.

Ordering the moves

When many partitions need to move, the order determines the peak concurrent load. Three policies are in use, and they trade recovery time against peak disruption.

All at once. Every partition starts migrating immediately. Recovery time is minimal if the network can carry the aggregate, and the cluster is unusable if it cannot. The peak concurrency is P, and the aggregate demand is P times the per-move rate, which for a large P exceeds the cluster’s bandwidth by orders of magnitude. Every move then takes longer, and the total recovery time is bounded by bandwidth rather than by concurrency, so the extra concurrency buys nothing.

Fixed concurrency. At most K moves run at a time, with K chosen so that K times the per-move rate fits within the cluster’s spare bandwidth. Recovery time is total bytes divided by the spare bandwidth, which is the same bound as all-at-once, but the peak disruption is controlled. K is the right parameter when the spare bandwidth is known and stable.

Priority order. Moves are ordered by the criticality of what they restore. A partition with no surviving replica is restored first, because it is unavailable; a partition with one surviving replica is restored next, because it is one failure from unavailable; a partition that is merely imbalanced is moved last. This ordering minimizes the time during which data is at risk, at the cost of delaying balance. It is the right policy after a failure, and the wrong policy for routine balancing.

The distinction between restoring replication and balancing load is important and often blurred. A rebalancer that treats both as “moves” will delay replication repair behind load balancing, which is backwards: an under-replicated partition is a durability risk, while an imbalanced one is a performance issue. Separating the two into different queues with different priorities is the standard fix.

Why the storm returns after the first one

A recovery that saturates the cluster also fills its queues with pending work: compacted files that need merging, index rebuilds that were deferred, caches that were evicted. When the migration finishes, the deferred work runs, and it competes with the restored request traffic. The cluster sees a second load spike that is not migration traffic and is therefore not throttled by the migration controller.

The second spike is often larger than the first, because it is the accumulated backlog of everything that was postponed. Its shape is a step, not a ramp, and it arrives exactly when the cluster is trying to return to steady state. A recovery plan that does not account for deferred maintenance will produce a two-peaked latency curve, and the second peak is the one that triggers the next failure.

The mitigation is to defer the deferred work too: compaction and index rebuilds should be scheduled by the same admission control as migration, with a rate that ramps up as the cluster’s latency returns to target. The controller is the same one; only the stream being throttled changes.

Measuring whether it worked

The metrics that matter are not the migration throughput. They are:

The time during which any partition has fewer replicas than its target. This is the durability exposure, and it should be the primary recovery metric. It is bounded below by the data that must be copied divided by the cluster’s spare bandwidth, and no policy improves on that bound.

The request latency during recovery, compared to the latency before the failure. A recovery that doubles the p99 has cost the cluster more than the failure itself, if the doubling lasts long enough. The acceptable ratio is a policy decision, and it is the input to the controller’s target.

The migration rate over time. A rate that starts high and decays indicates the controller is cutting back; a rate that oscillates indicates the controller’s gain is too high; a rate that stays at the minimum indicates the target is unreachable and the cluster is genuinely out of capacity.

The queue depth on the source nodes. A source with a growing queue is the leading indicator of the storm, and it moves before the latency metric does. Watching it gives the controller a head start that latency alone does not.

None of these are the rebalancer’s own success metric, which is usually “imbalance below threshold”. That metric can be satisfied by a rebalancer that has made the cluster unusable, and it is the reason the rebalancer’s threshold should never be the only thing being measured. The interaction between the rebalancer and the request path is the system, and the system is what has to be observed.