Even overall, lopsided per stream

A Felix stream is split into shards, and every shard has one leader: the broker that appends its records. Placement decides which broker that is, and for a long time it kept the cluster fair. Three brokers, 61 leaders, 21 on one, 21 on another, 19 on the third.

That looked even. For one stream it wasn't. Twelve shards of a single stream had landed 1, 1 and 10, and the broker with 10 was doing nearly all of that stream's work.

Figure 1Even by count, lopsided by stream. The totals read 21, 21 and 19; one stream's twelve leaders sit 1, 1 and 10.

Why one stream's spread matters

A write to a shard is served by the shard's leader. A cluster client sends each batch straight to the leader of the shard it belongs to. A plain client sends it to the broker it is connected to, and that broker forwards the batches for shards it doesn't lead. Either way the leader appends the records and replicates them.

So a stream's write capacity is the capacity of the brokers that lead its shards. Twelve shards on three brokers should give the stream three brokers' worth. Twelve shards led by one broker give it one, however many shards there are.

Figure 2A shard takes writes at its leader. The top row is one stream as it landed, 0, 0 and 12; the bottom row is the same stream spread 4, 4 and 4.

The cluster total doesn't show this. It adds every stream's leaders together, so a stream that is heavy on one broker can sit beside streams that are heavy on the other two and the sum still comes out level. A deployment dominated by one or two streams is the case that suffers: those streams get roughly a third of the cluster's write capacity.

What we saw

We found it on a benchmark of three brokers with replication factor 3 and 12 shards per stream. With other streams already filling two brokers to their share, a recreated 12-shard stream landed 0/0/12. Recreating it landed it 1/1/10, and three recreates in a row gave the same skew. The replication status listed the streams' leaders per broker, and for the stream in question it read perf-quorum:0=1,1=1,2=10. Across the cluster it came to 21/21/19, which is balanced.

The load followed the leaders. In the first measured cell, the broker with 10 leaders used 3.62 cores and the other two about 0.01.

The cause was in what placement counted. It tallied leaders per broker across the whole cluster and nothing else (leader_share and the leaders tally in plan.rs). A broker was "full" once it led ceil(shards / live brokers) of all shards. A new stream's shards went to brokers that weren't full, and when two brokers were full already, the third took everything. Rebalancing then stopped for the same reason: with totals within the share, no broker looked over.

Neither check knew which stream a shard belonged to.

Counting per stream

The fix is a second tally. placement/leadership.rs counts each broker's leaders in total and per stream or cache. For a stream with shards shards on nodes eligible brokers, placement works out two numbers beside the cluster share: a floor, shards / nodes rounded down, and a ceiling, rounded up. For twelve shards on three brokers both are 4.

Everything else uses those numbers. The cluster share is still there, and it still matters, but it has moved down a level: it decides between brokers that are equally good for the stream.

Placing a new shard

A new shard, or one whose leader failed over, goes to the highest-scoring broker that passes the first test it can. In order:

A broker under the stream's floor, preferring one under the cluster share.
A broker under the stream's ceiling, again preferring one under the cluster share.
Then the rules that existed before: a broker under the cluster share, then any broker with room, then the least loaded.

Figure 3Placing a new shard. The stream's floor and ceiling come first, and the cluster share only chooses between brokers that fit the stream.

Each broker gets its floor of the stream before any gets its ceiling, so when every broker has room, a stream ends no more than one shard apart wherever its shards land. A fresh 12-shard stream on three brokers is placed 4/4/4 whatever the other streams already lead, again when every broker has room, and it needs no moves afterwards.

Scores come from rendezvous hashing of the shard key and the broker id, with ties broken on the id, and plan walks the shards in sorted order. Two control-plane instances given the same snapshot therefore plan the same layout. The new tiers keep that: they only add filters in front of the same ordering.

Draining a broker and promoting a follower use the same idea. They prefer brokers under both shares and fall back as before. A promotion also counts toward its broker in the same pass, so the next shard in the pass sees it.

Moving the ones that are already placed

Placement also runs on a timer against streams that exist. For a stream already skewed, it has two goals, in this order.

The stream. The broker leading the most of a stream's shards, when that is at least two more than another broker leads, hands one to the broker leading the fewest of that stream.

The cluster. Once no stream wants a move, the same rule applies to the totals. The broker leading the most overall, when it is at least two ahead of another, hands over one shard. The shard must belong to a stream the destination leads fewer of, so evening the totals never makes a stream more uneven.

The order is deliberate. A shard moved for the totals while a stream still wanted its own moves could land somewhere the stream's move then took it away from again. Without it, a broker that joined was given shards that were later moved off again. In code, each pass runs once with cluster moves off, and runs again with them on only if no stream wanted a move.

Figure 4Moves go from the broker leading most to one leading at least two fewer, one stream's skew at a time. The totals wait until no stream wants a move.

Why it doesn't thrash

Every move for a stream lowers that stream's sum of squared counts, because it goes from a broker with at least two more to one with at least two fewer. Every move for the totals leaves each stream's sum the same or lower, and lowers the cluster's. So the pair (each stream's spread, the cluster's spread) goes down with every move, and placement's own moves can't undo each other. (A move you start by hand with POST /v1/shard-moves is yours to judge; placement will treat the result like any other layout.)

A cluster whose streams are lopsided is rebalanced the same way: one move at a time, until each stream is within its floor and ceiling and the totals are within one.

Counts include the moves already in flight as if they had finished, so a broker never stages more moves than it needs.

What keeps it safe

None of the limits around a move changed. A leader moves by staging a copy on the destination, fencing the old leader and cutting over, and the staged copy is a full copy of the shard's log. So the cluster runs one move at a time by default, with an optional limit on moves into or out of any one broker. Pausing placement starts none. A rebalance is optional, so it only goes where the shard keeps as many zones as it has now, and a broker with a max_shards cap is never filled past it.

Placement decides an assignment before it is proposed to the control plane's Raft group, and the entry carries the decided assignment. No member applies an entry differently, so nothing here needed a cluster-wide gate.

One thing it doesn't do: it counts shards, not bytes. A stream whose traffic is lopsided across its own shards is still lopsided. Spreading the leaders makes sure the shards that exist aren't piled on one broker.

What we checked

The tests are in tests/balance.rs, next to the placement code. The layout from the benchmark, four streams and a cache with the 12-shard stream at 1/1/10, ends 4/4/4, with every stream and the totals within one. Five streams skewed different ways balance per stream and in total, with no shard moving twice. A new 12-shard stream beside brokers at 6/6/0 is placed 4/4/4 and never moves. Further tests cover a joining broker, a draining one, a broker with a shard cap, zones, and a cluster of five brokers with replication factor 3.

We reverted the source change and kept the tests. The balance tests failed, including the joining-broker test on the first version of the rule. With the change restored, they passed.

Trying it

Placement is on by default; nothing needs configuring. Adding, draining and removing brokers describes what moves a shard and how to watch it, and Moving shards by hand covers pausing placement and starting a move yourself with POST /v1/shard-moves. The placement rules themselves are in the control plane design notes.

To check a stream of your own, list its shard assignments and count leaders per broker. Level means no broker leads more than one above the broker that leads fewest of the stream.