EvenScheduler: opt-in round-robin rebalance onto returning idle supervisors

[mwkang]

What is the purpose of the change

EvenScheduler (and therefore DefaultScheduler) does not move workers onto a supervisor that returns to service after maintenance. The topology's desired worker count is already satisfied across the surviving supervisors, so needsScheduling reports nothing to do and the returned supervisor sits at used = 0 until an operator manually rebalances or restarts every affected topology.

This PR adds an opt-in, binary-trigger pass to EvenScheduler that relocates already-assigned workers onto such idle supervisors, round-robin across topologies, in a single scheduling round. It is disabled by default, so existing clusters see no behavior change. Implements the proposal in #8590 and folds in the review feedback from that thread.

How it works

The trigger and the relocation live entirely on the EvenScheduler path; Cluster.needsScheduling is intentionally left unchanged (see Scope below).

1. Binary trigger — Cluster.hasIdleSupervisorReusableBy(topology) returns true only when at least one stable, non-blacklisted supervisor has zero used slots and the topology is not already on it. Because the check is binary (a supervisor either has zero used slots or it does not), it never fires for an "almost balanced" cluster, so no time-based cooldown is needed.
2. Per-topology budget — each topology may relocate at most floor(numWorkers / nonBlacklistedSupervisorCount) * idleSupervisorCount workers per round, tightened further by nimbus.even.rebalance.max.free.per.topology when positive. A topology whose budget computes to 0 (typically numWorkers < supervisorCount) is skipped entirely — this is also what stops a single-worker topology from ping-ponging.
3. Round-robin relocation — EvenScheduler.redistributeOntoIdleSupervisors walks the eligible topologies (ordered by id) and moves at most one worker per topology per iteration until the idle slots are exhausted. A single returning supervisor therefore ends up hosting workers from several topologies, preserving the per-supervisor workload diversity a fresh submission has, instead of letting the first scheduled topology grab the entire idle capacity.
4. Deterministic donor selection — a worker is pulled from the supervisor where the topology currently has the most workers (measured by worker count), ties broken by supervisor id, lexicographically. The source supervisor is never drained below one worker for that topology, so it cannot itself become the next round's idle supervisor.
5. Direct placement — each pulled worker's executors are assigned directly onto an idle slot via cluster.freeSlot() + cluster.assign(), bypassing the regular sortSlots/interleave pass that would otherwise drop some of them straight back into the just-vacated slots.

Nimbus propagates the result the usual way: it diffs the resulting assignments against the existing ones and pushes the delta, so the relocation takes effect even though needsScheduling is untouched.

Scope: RAS, Multitenant, Isolation

The feature is scoped to EvenScheduler/DefaultScheduler (and the leftover topologies IsolationScheduler delegates to them). Cluster.needsScheduling is deliberately not modified — the new logic lives in three new Cluster methods (hasIdleSupervisorReusableBy, isIdleSupervisorAvailableForEvenRebalance, hasMinimumIdleSupervisorStability) reached only from EvenScheduler.redistributeOntoIdleSupervisors. This keeps any scheduler that consults needsScheduling from picking up a surprise "needs rescheduling" signal.

Call-path audit:

Caller	Trigger it uses	Reaches the new idle-rebalance path?
EvenScheduler.scheduleTopologiesEvenly	needsScheduling (unchanged)	Yes — calls redistributeOntoIdleSupervisors directly; gated by the default-off flag
DefaultScheduler.defaultSchedule	needsScheduling (unchanged)	Yes — same, gated by the flag
IsolationScheduler	delegates leftover (non-isolated) topologies to DefaultScheduler.defaultSchedule	Reached but neutralized — isolated hosts are blacklisted, so they are never a donor or a target
ResourceAwareScheduler	needsSchedulingRas (unchanged)	No — never calls Even/Default scheduler
MultitenantScheduler (DefaultPool / IsolatedPool)	calls cluster.needsScheduling (unchanged)	No — uses its own pools and never reaches redistribute; needsScheduling itself is unmodified

In words: RAS is intentionally out of scope — it uses needsSchedulingRas and a different placement engine; a parallel mechanism, if wanted, belongs in a follow-up. Multitenant pools do call needsScheduling, but since that method is unchanged they are unaffected. Isolation: both hasIdleSupervisorReusableBy and redistributeOntoIdleSupervisors skip blacklisted supervisors, and IsolationScheduler represents a reserved host by blacklisting it before delegating leftovers — so an isolated host can never be a donor or a target, including the case where its isolated topology is down and the reserved host looks idle.

Configuration

All keys are dot-only, matching Storm's convention.

Key	Type	Default	Purpose
nimbus.even.rebalance.idle.supervisor.enabled	boolean	false	Master switch (opt-in)
nimbus.even.rebalance.max.free.per.topology	int	0	Optional per-topology upper bound per round (0 = unbounded; the even-distribution budget applies)
nimbus.even.rebalance.idle.supervisor.min.stable.rounds	int	3	Flap guard; 0 disables the guard

The flap guard keeps workers off a supervisor that has only just returned and may still be flapping on a slow JVM startup or a transient network blip. A supervisor is eligible only once it has been up for at least min.stable.rounds * supervisor.monitor.frequency.secs (≈9s with the defaults). It reuses SupervisorInfo.uptime_secs, surfaced onto SupervisorDetails.

When you would NOT want to enable this

A relocation is a worker JVM restart: brief tuple replay, JIT re-warmup, and possible windowed/stateful bolt state churn. Keep this off for:

• topologies with windowed or stateful bolts that pay a non-trivial replay/restore cost;
• latency-sensitive topologies sensitive to JIT re-warmup;
• clusters whose supervisors flap (raise min.stable.rounds, or leave the feature off);
• RAS users (no effect — see Scope).

Blast radius

In one scheduling pass the simultaneous worker-restart count is min(idle_slots, eligible_topologies), with each topology's contribution capped at floor(numWorkers / nonBlacklistedSupervisorCount) * idleSupervisorCount (tightened by max.free.per.topology). Because every relocation consumes one idle slot, the total per pass is hard-bounded by the returning supervisor's free-slot count.

Worked example: one returning supervisor with 8 slots in a 50-topology cluster → 8 simultaneous worker restarts across 8 topologies in one pass, not 50.

A cluster-wide ceiling (nimbus.even.rebalance.max.relocations.per.round) was considered but not added: the per-topology cap plus the natural idle-slot ceiling already bound the disruption, and an extra knob would only let operators throttle below "fill the returned supervisor in one pass." Happy to add it if reviewers prefer an explicit cluster-wide cap.

How was this change tested

New TestEvenSchedulerIdleSupervisor (storm-server), 16 cases:

• disabled-by-default no-op and the binary trigger;
• the generic needsScheduling / needsSchedulingRas paths staying unaffected;
• the per-topology drain cap and max.free.per.topology;
• drain-to-zero protection and the single-worker no-op;
• one-round even distribution and round-robin sharing across topologies;
• the uptime flap guard (below-threshold → no move, at-threshold → move);
• deterministic donor tie-break by supervisor id;
• blacklisted idle supervisor excluded as a target;
• the DefaultScheduler leftover-subset path;
• the IsolationScheduler interaction (idle non-isolated target only; reserved host stays out even when its isolated topology is down).

Backward compatibility

Default-off with no API removals — only additions. When disabled, redistributeOntoIdleSupervisors returns before scanning any supervisor, so a cluster that has not opted in does no extra per-round work. Existing SupervisorDetails constructors default uptimeSecs to Long.MAX_VALUE (always "stable"), leaving every existing caller unchanged.

Closes #8590

[reiabreu]

Disclaimer: ran this through an LLM for thoroughness

Most of the following items are nice-to-haves, but can you check numbers 4 and 5?
Thank you

1. 🟡 Medium — Missing comment on the null-guard in DefaultScheduler / EvenScheduler

for (TopologyDetails topology : cluster.needsSchedulingTopologies()) {
    if (topologies.getById(topology.getId()) == null) {
        continue;
    }

This guard is non-obvious — it exists because redistributeOntoIdleSupervisors can call
freeSlot + assign on a topology that was previously "needs scheduling", which causes its
snapshot to disappear from Topologies mid-loop. Worth a one-line comment so future
contributors don't remove it as dead code.

2. 🟢 Low — hasIdleSupervisorReusableBy re-reads config and re-iterates supervisors on every call

redistributeOntoIdleSupervisors already builds the idle-supervisor set and checks the
feature flag. hasIdleSupervisorReusableBy then re-evaluates both for every topology in the
per-topology budget loop, creating O(topologies × supervisors) redundant work. It's
negligible at scale, but a simple cached boolean enabled and a pre-computed idle-supervisor
set passed into the topology loop would avoid the duplication entirely.

3. 🟢 Low — SupervisorDetails constructor count

The class goes from 6 to 10 constructors. This is consistent with the existing style, but a
static factory pair — SupervisorDetails.of(...) / SupervisorDetails.withUptime(...) — or
a builder would make it easier to see which fields are being set. Not a blocker, just worth
considering given 3.0.0 is a major release that can break APIs.

---

Tests

4. 🔴 High — Missing total-worker count assertions

Several tests verify per-supervisor slot counts but don't assert that the total number of
assigned workers is preserved. A concrete example of what slips through:

In redistributeRelocatesAtMostMaxFreeWorkersPerTopology the test asserts:

assertEquals(1, usedSlotCount(cluster, "sup-0"));
assertEquals(1, usedSlotCount(cluster, "sup-1"));
assertEquals(1, usedSlotCount(cluster, "sup-2"));

Imagine a bug in relocateOneWorkerOntoIdleSlot where cluster.assign(target, ...) is
called but the executor collection passed is empty (e.g. an erroneous execs == null early
return leaves the target slot occupied but the executors unassigned). All three
usedSlotCount assertions still pass — 3 slots are in use — but the topology now has 3
workers with fewer executors than it declared. The bug is invisible.

Adding one line to every relocating test closes this entirely:

// topology declared numWorkers=3; relocation must preserve that
assertEquals(3, cluster.getAssignedNumWorkers(firstTopology(cluster)));

Affected tests: redistributeRelocatesAtMostMaxFreeWorkersPerTopology,
scheduleTopologiesEvenly_movesOneWorkerToIdleSupervisor,
evenDistributionInOneRound_unboundedMaxFree, and the two scheduler-integration tests.

5. 🔴 High — Round-robin test under-asserts

multipleTopologies_shareIdleSlotsRoundRobin is the only test that exercises the core
round-robin fairness guarantee, but its assertions don't actually enforce it. The test
currently checks something like:

assertTrue(slotsOnSup2ForTopoA >= 1);
assertTrue(slotsOnSup2ForTopoB >= 1);

Consider a broken implementation where the inner topology loop exits after the first
successful topology instead of continuing round-robin. topo-A (sorted first by id) would
consume all 4 idle slots on sup-2; topo-B gets zero. Yet if sup-2 also happens to have
topo-A's workers already placed there via a previous iteration, the >= 1 check still
passes — the round-robin bug is completely invisible.

The fix is to assert exact counts and verify worker totals per topology:

// sup-2 has 4 slots; with 2 topologies, round-robin gives each exactly 1 slot
// (each topology's budget = floor(4 workers / 3 supervisors) * 1 idle = 1)
assertEquals(1, slotsOnSup2ForTopoA);
assertEquals(1, slotsOnSup2ForTopoB);

// Total worker count must be preserved for both topologies
assertEquals(4, cluster.getAssignedNumWorkers(topoA));
assertEquals(4, cluster.getAssignedNumWorkers(topoB));

// No supervisor should have been drained to zero
assertTrue(usedSlotCount(cluster, "sup-0") > 0);
assertTrue(usedSlotCount(cluster, "sup-1") > 0);

Without these, the test is really only checking "something happened on sup-2", not that
round-robin sharing works.
6. 🟡 Medium — Flap-guard boundary coverage

Tests flapGuard_belowThreshold_doesNotMove and flapGuard_atThreshold_moves cover
uptime < threshold and uptime == threshold as two separate methods. A @ParameterizedTest
with threshold - 1, threshold, and threshold + 1 would give stronger boundary coverage
with less code and make the off-by-one contract explicit.

7. 🟢 Low — noIdleSupervisor_doesNotTrigger conf inconsistency

This test hand-builds its conf map and omits NIMBUS_EVEN_REBALANCE_IDLE_SUPERVISOR_MIN_STABLE_ROUNDS,
while every other test goes through evenRebalanceConf(...). The test passes because
genSupervisors defaults to Long.MAX_VALUE uptime, but the inconsistency makes it harder
to trust at a glance. Routing it through the shared helper would make the intent clearer.

8. 🟢 Low — Setup duplication

noIdleSupervisor_doesNotTrigger, redistributeNeverDrainsSupervisorToZero,
singleWorkerTopology_doesNotMoveDespiteIdleSupervisors, and
evenDistributionInOneRound_unboundedMaxFree each inline 15–25 lines of slot/executor
wiring that is almost identical to buildClusterWithIdleSupervisor. Extracting a small
fluent ClusterBuilder test helper (or at least a shared buildAssignment(topology, slots[])
method — which multipleTopologies_shareIdleSlotsRoundRobin already has!) would halve the
boilerplate.

[mwkang]

Thanks for the thorough review — and for flagging 4 and 5 specifically; those were the right things to tighten. I've pushed a follow-up addressing the feedback.
Item by item:

4 (total-worker assertions) — Done. Every relocating test now asserts cluster.getAssignedNumWorkers(topology) against the declared numWorkers: redistributeRelocatesAtMostMaxFreeWorkersPerTopology, scheduleTopologiesEvenly_movesOneWorkerToIdleSupervisor, evenDistributionInOneRound_unboundedMaxFree, and both scheduler-integration tests — plus donorTieBreaksBySupervisorIdWhenWorkerCountsTie, which was the one relocating test I'd missed. This is a genuinely independent check: getAssignedNumWorkers counts distinct slots in the executor→slot map, whereas usedSlotCount reads nodeToUsedSlotsCache, so an assign(target, <empty execs>) bug now fails the count assertion even though the slot still registers as used.

5 (round-robin under-asserts) — Done. multipleTopologies_shareIdleSlotsRoundRobin now asserts exact counts rather than >= 1: usedSlotCount(sup-2) == 2 with supervisorWorkerCount == 1 for each of topo-A and topo-B, getAssignedNumWorkers == 4 for both, and that neither donor (sup-0/sup-1) is drained to zero. I confirmed the new assertions actually catch the failure mode by temporarily breaking the inner loop so the first topology grabs both idle slots — the test then fails on topo-B's count, as intended.

1 (null-guard comment) — Added a comment above the guard in both DefaultScheduler and EvenScheduler. The rationale turned out to be slightly different from the one in your note: needsSchedulingTopologies() returns the cluster's full topology set, while each caller scopes the run to the topologies it passed in (the full set from *.schedule, where the guard is a no-op; a single-topology Topologies per leftover from DefaultScheduler.defaultSchedule; the leftover non-isolated subset from IsolationScheduler). The guard skips anything outside that set so the leftover path never schedules a topology the caller excluded — e.g. a down isolated topology on a reserved host. The comment now spells out those cases.

2 (redundant config re-read / supervisor rescan) — Done. The feature-flag check is hoisted to the top of redistributeOntoIdleSupervisors (early return), and the idle-supervisor set is computed once up front and passed into a new private topologyCanReuseIdleSupervisor(...) helper, replacing the per-topology cluster.hasIdleSupervisorReusableBy(...) call. It's behaviorally equivalent — same getUsedSlotsByTopologyId existence check against the idle set — but the config read and full supervisor scan now happen once per round instead of once per topology. (Cluster.hasIdleSupervisorReusableBy stays as the public binary-trigger oracle the tests assert against.)

6 (flap-guard boundary) — Done. The two methods are merged into a single @ParameterizedTest (flapGuardHonorsMinStableRoundBoundary) with @CsvSource covering threshold-1 / threshold / threshold+1 (8→no-move, 9→move, 10→move at minStableRounds=3, monitorFreq=3), asserting both the trigger boolean and the resulting placement. Makes the uptimeSecs >= minStableRounds * monitorFrequencySecs contract explicit.

7 (conf inconsistency) — Done. noIdleSupervisor_doesNotTrigger now goes through the shared evenRebalanceConf(...) / newCluster(...) helpers, and I swept up the last remaining hand-built fixture (multipleTopologies_shareIdleSlotsRoundRobin) so every test uses the same path.

8 (setup duplication) — Done. Extracted a shared buildAssignment(topology, slots[]) helper; the four tests you listed now use it together with newCluster/evenRebalanceConf, which removes the bulk of the inline slot/executor wiring.

3 (SupervisorDetails constructor count) — Left as constructors for now. The new uptimeSecs overloads chain through to existing defaults (Long.MAX_VALUE), so every current caller is unchanged, and the style matches the rest of the class. A static factory / builder is a reasonable cleanup, but it's a public-API change that's somewhat orthogonal to this PR — I'd prefer to keep it out of scope here and do it as a separate follow-up if you feel it's worth it. Let me know.

Thanks again!

[rzo1]

Nice, thanks for the PR! A few non-blocking items.

1. Cluster.hasIdleSupervisorReusableBy(...) is dead in production. The scheduling path uses the private topologyCanReuseIdleSupervisor(...) instead (with a comment noting it's "equivalent … but reuses the pre-computed set"), so the public Cluster method is reached only by the tests and the description's call-path table. A public API on a core class kept alive solely for assertions is a smell, and the two implementations can drift. Either route the production trigger through it, or drop it and assert against redistributeOntoIdleSupervisors' observable effect.

2. Two opposite "unknown uptime" defaults. The legacy SupervisorDetails constructors default uptimeSecs to Long.MAX_VALUE (always eligible), while Nimbus.supervisorUptimeSecs() returns 0L when unset (never eligible). Production always takes the 0L branch — conservative, good — so Long.MAX_VALUE only affects callers that don't use this feature. Defensible, but the opposite defaults surprise; one short comment on each would help the next reader.

3. Budget is based on declared workers, not assigned. target = (topo.getNumWorkers() / nonBlacklistedSupervisorCount) * idleSupervisorCount uses declared numWorkers, so for an under-assigned topology it over-estimates. relocateOneWorkerOntoIdleSlot returns false once no donor has ≥2 workers, so it self-limits and can't over-move — but a one-line javadoc noting the cap is an upper bound, not a guarantee, would be accurate.

4. freeSlot then assign is non-atomic. If assign ever threw after freeSlot(victim), the executors are orphaned for that round (recovered by the normal pass). Target is a verified idle free slot so this is near-impossible in practice — a comment noting the ordering assumption is enough.

5. Minor: @VisibleForTesting on redistributeOntoIdleSupervisors reads as "would be private," but it's genuinely the cross-class entry point (called from DefaultScheduler and EvenScheduler). Slightly misleading annotation.

Test coverage

Two parts of the budget formula aren't actually exercised by the current cases:

• The idleSupervisorCount multiplier always resolves to * 1. Every test has exactly one usable idle supervisor (sup-2; the reserved-host test blacklists sup-2, leaving only sup-3), so a regression in the * idleSupervisorCount term would go unnoticed. A case with two simultaneously-idle supervisors would cover it.
• The max.free.per.topology clamp is never the binding constraint. In redistributeRelocatesAtMostMaxFreeWorkersPerTopology the even-budget is floor(3/3)*1 = 1 and maxFree = 1, so the two coincide and the test still passes even if Math.min(target, maxFree) were removed. A fixture where the even-budget is ≥2 but maxFree = 1 would pin down the clamp specifically.

Overall LGTM once #1 is resolved one way or the other. 👍