[PROPOSAL] EvenScheduler: opt-in round-robin rebalance to populate returning idle supervisors

[mwkang]

Summary

Currently EvenScheduler (and therefore DefaultScheduler) does not move workers onto a fully-idle supervisor that returns to service after maintenance. The cluster keeps running with used slot = 0 on the returned supervisor until an operator manually rebalances or restarts every affected topology.

This proposal introduces an opt-in, binary-trigger rebalance pass inside EvenScheduler. When at least one non-blacklisted supervisor has zero used slots, the scheduler relocates one worker per topology per round-robin iteration onto the idle slots, until the idle capacity reaches an even per-supervisor share. Already-balanced topologies are never touched.

The feature is disabled by default so existing behavior is preserved.

---

Problem statement

Today's Cluster#needsScheduling returns true only when the topology has fewer assigned workers than desired or has unassigned executors:

public boolean needsScheduling(TopologyDetails topology) {
    int desiredNumWorkers = topology.getNumWorkers();
    int assignedNumWorkers = this.getAssignedNumWorkers(topology);
    return desiredNumWorkers > assignedNumWorkers
        || getUnassignedExecutors(topology).size() > 0;
}

When a supervisor goes down for maintenance and comes back, the topology's workers were already redistributed onto the surviving supervisors, so both conditions are false and the scheduler considers the cluster "fully scheduled". The returning supervisor sits idle indefinitely.

Initial state                  After supervisor maintenance
sup-A    sup-B    sup-C        sup-A    sup-B    sup-C
+-+-+-+  +-+-+-+  +-+-+-+      +-+-+-+  +-+-+-+
|W|W| |  |W|W| |  |W|W| |      |W|W|W|  |W|W|W|       (sup-C is down)
+-+-+-+  +-+-+-+  +-+-+-+      +-+-+-+  +-+-+-+

After sup-C returns                  Today's behavior
sup-A    sup-B    sup-C              sup-A    sup-B    sup-C
+-+-+-+  +-+-+-+  +-+-+-+            +-+-+-+  +-+-+-+  +-+-+-+
|W|W|W|  |W|W|W|  | | | |    -->     |W|W|W|  |W|W|W|  | | | |
+-+-+-+  +-+-+-+  +-+-+-+            +-+-+-+  +-+-+-+  +-+-+-+
                                                       (stays idle forever)

Operators must currently restart each affected topology by hand to drain the over-loaded supervisors.

---

Proposed approach

1. Binary trigger

Cluster#needsScheduling gets one extra branch: also return true when at least one non-blacklisted supervisor has zero used slots and the topology is not already on that supervisor. The check is binary — a supervisor either has zero used slots or it does not — so a near-balanced cluster never triggers this path.

Triggers       sup-A    sup-B    sup-C
               +-+-+-+  +-+-+-+  +-+-+-+
               |W|W|W|W|W|W|W|W| | | | |   <-- triggers
               +-+-+-+  +-+-+-+  +-+-+-+

Does not       sup-A    sup-B    sup-C
trigger        +-+-+-+  +-+-+-+  +-+-+-+
               |W|W|W|W|W|W|W| | |W| | |   <-- (4, 3, 1) is not balanced
               +-+-+-+  +-+-+-+  +-+-+-+       but sup-C has used > 0
                                                so trigger does not fire

2. Round-robin relocation across topologies

A new phase at the start of scheduleTopologiesEvenly walks idle slots round-robin across all triggering topologies. Each iteration moves at most one worker per topology, so the returning supervisor ends up hosting workers from many topologies — preserving the per-supervisor workload diversity that a fresh submission produces.

Topologies A, B, C, D each at distribution (4, 4, 0).
sup-C just returned with 4 free slots.

Round-robin iterations:
  iter 1: A's worker -> sup-C:0
  iter 2: B's worker -> sup-C:1
  iter 3: C's worker -> sup-C:2
  iter 4: D's worker -> sup-C:3
  done (idle capacity exhausted).

Result on sup-C: { worker of A, worker of B, worker of C, worker of D }

If only one topology triggers, that topology fills the idle supervisor up to its even share and stops. If only some topologies are eligible, the remaining idle slots stay empty for this round.

3. Per-topology budget per round

Each topology contributes at most
idleSupervisorCount * floor(numWorkers / nonBlacklistedSupervisorCount)
workers in one scheduling round, capped further by the idle side's free slot capacity. This means:

• topologies with numWorkers < numSupervisors automatically skip the round-robin (their floor is zero) — single-worker topologies cannot ping-pong;
• "near balanced" cluster never triggers in the first place;
• one round produces an even per-supervisor share and the next round's trigger is silent.

4. Direct placement onto idle slots

The relocated executors are assigned directly onto idle slots via cluster.freeSlot(victim) + cluster.assign(idleSlot, ...). This bypasses the regular sortSlots/interleaveAll pass that would otherwise be free to drop some of the freed executors back into the just-vacated slots on the busy supervisors.

---

Configuration keys

Two new entries in DaemonConfig:

Key	Type	Default	Meaning
nimbus.even.rebalance.idle-supervisor.enabled	boolean	false	Master switch. When false, the new code paths are short-circuited and behavior is identical to today.
nimbus.even.rebalance.max-free-per-topology	int	0	Optional upper bound on the number of workers a single topology may release per round. 0 or negative means unbounded (the per-round budget above takes effect).

---

Safety guards (already in the draft)

• Opt-in (default off): zero behavior change for existing deployments.
• Binary trigger: an "almost balanced" cluster cannot trigger a relocation; no time-based cooldown is needed because the trigger condition itself rules out instability.
• Drain-to-zero protection: a supervisor that holds only one worker of a topology is never the donor — the same round can never produce a new idle supervisor.
• Round-robin across topologies: prevents the first-scheduled topology from monopolizing the idle capacity, which would concentrate one workload on the returning supervisor.
• Idle-side slot cap: the round-robin pass never tries to free more workers than the idle capacity can absorb.
• Direct placement: relocated executors never bounce back onto the just-vacated busy supervisors.

---

Implementation outline

The draft (against master / 3.0.0-SNAPSHOT) touches three production files plus a test:

storm-server/src/main/java/org/apache/storm/DaemonConfig.java
   - Two new keys (above).

storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java
   - needsScheduling: extra branch via hasIdleSupervisorReusableBy().
   - hasIdleSupervisorReusableBy(): binary check, returns false when disabled.

storm-server/src/main/java/org/apache/storm/scheduler/EvenScheduler.java
   - scheduleTopologiesEvenly(): calls redistributeOntoIdleSupervisors() first.
   - redistributeOntoIdleSupervisors(): round-robin across topologies.
   - relocateOneWorkerOntoIdleSlot(): pulls one worker from the most-loaded
     supervisor of a topology, places it directly onto an idle slot.

conf/defaults.yaml
   - Defaults for the two new keys.

storm-server/src/test/java/.../TestEvenSchedulerIdleSupervisor.java
   - 9 unit tests covering trigger, drain cap, single-worker no-op,
     drain-to-zero protection, one-round even distribution, and
     round-robin sharing across multiple topologies.

I will open the PR once this proposal direction is agreed upon.

---

Backward compatibility

• Default off, so out-of-the-box behavior of DefaultScheduler / EvenScheduler is byte-for-byte identical.
• No public API removed or renamed. Two new public methods (Cluster#hasIdleSupervisorReusableBy, EvenScheduler#redistributeOntoIdleSupervisors) are additions only.
• New config keys both have safe defaults; no new YAML is required for existing clusters.

Feedback welcome on direction, naming, and scope before I open the PR. Thanks!

[rzo1]

Thanks @mwkang — well-structured proposal, the diagrams make the problem and the fix immediately legible. Default-off opt-in plus the binary trigger argument (no cooldown needed because "almost balanced" can't fire it) is the right framing, and the round-robin-across-topologies choice is a nice touch. A handful of things I'd want clarified or expanded before this moves to a PR:

1. The proposal is scoped to EvenScheduler / DefaultScheduler, but the de-facto modern scheduler in Storm is ResourceAwareScheduler, which uses needsSchedulingRas and an entirely different placement engine. Most non-trivial deployments run RAS, so it'd be good to state explicitly whether (a) RAS users are intentionally out of scope for this change, or (b) a parallel mechanism is envisioned later. Either is fine — but the doc should not leave it ambiguous.

2. Worth being explicit that a "relocation" is a worker JVM restart: brief tuple replay, JIT warmup, possible windowed/stateful bolt churn. The proposal frames the trigger purely in balance terms, but the cost side is real and is exactly why some operators will choose to keep this off. A short "when you would not want to enable this" note would help operators decide.

3. Interaction with IsolationScheduler / MultitenantScheduler is worth verifying explicitly: isolated topologies should never be selected as donors, and a supervisor reserved for an isolation slot shouldn't be considered "idle" just because the isolated topology happens to be down. A test for each case would be reassuring.

4. "Most-loaded supervisor of a topology" — by what measure (worker count vs. executor count) and what's the tie-break when two supervisors are equally loaded? Determinism matters for operators trying to understand why a particular worker moved. The PR should specify it and tests should pin it.

5. The binary-trigger argument convincingly rules out thrash, but a supervisor returning from maintenance can still flap on a slow JVM startup or a transient network blip. Moving workers onto a node that disappears 30s later doubles the disruption. A small "supervisor must have been alive for N scheduling rounds before being considered idle-and-available" guard would be cheap insurance — happy to see it as either a third config knob or a fixed conservative default.

6. needsScheduling is on a broadly-called path — invoked from needsSchedulingTopologies and indirectly by callers that aren't EvenScheduler-aware. Probably fine because RAS uses needsSchedulingRas, but worth confirming in the PR description that no other scheduler picks up a surprise "needs rescheduling" signal from the new branch.

7. nimbus.even.rebalance.idle-supervisor.enabled mixes dots and dashes. Storm convention is dots only (e.g. nimbus.even.rebalance.idle.supervisor.enabled). Minor, but easier to fix now than after the key is documented.

8. "One worker per topology per round-robin iteration" describes the fairness rule, but the aggregate simultaneous disruption is min(idle_slots, eligible_topologies) worker restarts in one scheduling pass. On a "one returning supervisor with 8 slots, 50 topologies" cluster that's 8 simultaneous worker restarts across 8 topologies. Worth quantifying so operators understand the realistic blast radius before they opt in.

Overall take: the core idea is sound, the safety guards are well-chosen, and the API/config story is appropriately conservative. I'd encourage you to open the PR. The asks before merge from my side would be: explicit scoping wrt RAS, an honest disruption note in the docs, isolation/multitenant interaction tests, and the smaller additions (grace period, deterministic tie-break, config-key naming). The binary-trigger argument and the round-robin policy are the strongest parts of the design — please keep both front-and-center in the PR description so reviewers don't re-litigate them.

Only my pov, but others might jump in too.

[mwkang]

Thanks @rzo1 — really appreciate the careful read. The eight points are all fair and several of them catch things I would have shipped as-is. Quick inline replies below; I'll fold the doc / PR-description items into the PR write-up and the code items into the patch itself.

1. RAS scope

Intentionally out of scope for this change. ResourceAwareScheduler calls Cluster.needsSchedulingRas (a separate method), and the new opt-in branch lives in Cluster.needsScheduling, so RAS keeps its existing trigger and never enters EvenScheduler.scheduleTopologiesEvenly. A parallel mechanism for RAS would belong in a follow-up — the placement engine is different enough that it warrants its own design. I'll state this explicitly in the PR description so RAS operators aren't left guessing.

2. "Relocation = worker JVM restart" note

Agreed, this is missing. I'll add a "When you would NOT want to enable this" section to the docs covering:

• topologies with windowed / stateful bolts that pay non-trivial replay cost,
• topologies sensitive to JIT warmup,
• clusters whose supervisors flap (see (5)),
• RAS users (no effect, see (1)).

The proposal sells the upside; the docs should sell the downside just as clearly.

3. Isolation / Multitenant interaction

Already safe, but I agree it should be explicit and tested.

• IsolationScheduler: blacklists isolated hosts before delegating leftover topologies to DefaultScheduler.defaultSchedule, which calls EvenScheduler.scheduleTopologiesEvenly. Both hasIdleSupervisorReusableBy and redistributeOntoIdleSupervisors skip blacklisted supervisors, so an isolated host can never be a donor or a target. The "isolated topology is down → host looks idle" case is covered by the same blacklist check.
• MultitenantScheduler doesn't route through EvenScheduler.scheduleTopologiesEvenly at all (it uses its own DefaultPool / IsolatedPool), so the new branch is unreachable from it.

I'll add two explicit tests:

• IsolationScheduler + idle non-isolated supervisor → relocation only outside the isolated set.
• IsolationScheduler with the isolated topology down → the isolation-reserved host is not treated as idle.

4. "Most-loaded supervisor" — measure and tie-break

Today:

• Measure: worker count per supervisor for that topology (not executor count). Workers are the unit being relocated, so worker count is the natural metric.
• Tie-break: currently relies on HashMap iteration order — which is not deterministic. That's a real operator-observability bug, good catch.

Fix: deterministic secondary key (supervisor id, lexicographic) with a test pinning the choice. PR description will spell out the rule so operators can reason about which worker moved and why.

5. Flap guard

Convincing. The binary trigger blocks thrash, but not the case where a supervisor returns, absorbs workers, and then disappears 30s later. I'll add a third opt-in knob:

nimbus.even.rebalance.idle.supervisor.min.stable.rounds  (default: 3)

A supervisor that has been present-and-idle for fewer than N consecutive scheduling rounds will not be considered eligible. Gated independently of the master switch so operators can tune it without re-evaluating the whole feature.

Implementation note: SupervisorInfo.uptime_secs is already available on the thrift side, so the cleanest path is to surface it on SupervisorDetails and gate hasIdleSupervisorReusableBy on uptime >= minStableRounds * supervisor.monitor.frequency.secs. No new state machine needed.

6. needsScheduling call-path audit

Verified across the tree:

Caller	Method used	Affected?
DefaultScheduler.defaultSchedule	needsSchedulingTopologies → needsScheduling	Yes, but gated by the (default-off) flag
EvenScheduler.scheduleTopologies	needsSchedulingTopologies → needsScheduling	Yes, same gating
ResourceAwareScheduler	needsSchedulingRas	No — separate method
IsolationScheduler	Indirect, via leftover topologies through DefaultScheduler.defaultSchedule	Reached but neutralized — blacklist masks the isolated set (see (3))
MultitenantScheduler	Doesn't go through needsScheduling	No

I'll include this table in the PR description.

7. Config key naming

Will fix. Renaming to dot-only:

• nimbus.even.rebalance.idle-supervisor.enabled → nimbus.even.rebalance.idle.supervisor.enabled
• nimbus.even.rebalance.max-free-per-topology → nimbus.even.rebalance.max.free.per.topology
• nimbus.even.rebalance.idle.supervisor.min.stable.rounds (new, from (5))

For reference, conf/defaults.yaml has ~280 dotted-only keys versus a single dash-containing one (storm.auth.simple-white-list.users), so dot-only is clearly the house style. Easier to fix now, before docs and release notes lock it in.

8. Blast radius

Your reading is correct. In one scheduling pass the simultaneous restart count is

min(idle_slots, eligible_topologies)

with the per-topology contribution capped by

floor(numWorkers / nonBlacklistedSupervisorCount) * idleSupervisorCount

(further tightened by max.free.per.topology when set to a positive value).

I'll lift your worked example into the docs verbatim: "one returning supervisor with 8 slots in a 50-topology cluster → 8 simultaneous worker restarts across 8 topologies in one pass." That's the number operators need to see before opting in.

Happy to add a cluster-wide ceiling on top of the per-topology one — something like nimbus.even.rebalance.max.relocations.per.round — if reviewers want a hard cap independent of topology shape. Let me know your preference; I'd lean toward adding it (cheap, conservative-by-default) but don't want to bolt on knobs that aren't pulling weight.

---

Plan from here: I'll fold (2), (4), (5), (7), (8), the audit table from (6), and the two new tests from (3) into the PR and open it. Points (1) and (3) need only doc/description framing — no code change. Will ping you on the PR once it's up. Thanks again.