[AMORO-4271] AIP-5 Phase 2: demand-driven scale-up for dynamic allocation groups

[j1wonpark]

Why are the changes needed?

Close #4271.

This is the second implementation phase of AIP-5: Dynamic Resource Allocation for Optimizer (#4191), building on the configuration foundation merged in #4254. It adds demand-driven scale-up for dynamic-allocation-enabled groups; scale-down lands in later phases. Everything remains opt-in: groups without dynamic-allocation.enabled = true are unaffected.

Why a dedicated keeper. The existing OptimizerGroupKeeper checks a group at optimizer-group.min-parallelism-check-interval × consecutive attempts — minutes apart by design, which is right for its floor duty but makes the DRA backlog timeouts (scheduler-backlog-timeout 1min, sustained-backlog-timeout 30s) physically unreachable. OptimizerScaleKeeper reuses the same AbstractKeeper infrastructure (HA leader gating, delay queue, lifecycle) but each group's scale evaluations run at its own sustained-backlog-timeout. The legacy keeper skips DRA groups (continuing the implicit isEffectivelyEnabled pattern from the Phase 1 review) while keeping them under watch, so it resumes floor duty if DRA is disabled later.

Why a "future demand" signal. Planning is only driven by pollTask: with zero optimizers nothing polls, planning never runs, and PLANNED tasks never materialize — so a task-based signal alone can never wake a scaled-to-zero group. PENDING tables are the one signal observable without optimizers; they ignite a single probe instance, and once planning materializes tasks, the task-based signal takes over.

Brief change log

• OptimizerProperties / DynamicAllocationConfig: new dynamic-allocation.executor-parallelism (default 1) — the scaling unit is one homogeneous K-thread optimizer instance (the Spark executor model). Validation: 1 ≤ executor-parallelism ≤ max-parallelism, and the floor must be reachable in K units (ceil(min/K)×K ≤ max) — otherwise the group would silently sit below its floor forever.

• OptimizerScaleKeeper (new, in DefaultOptimizingService): owns both the floor and the demand scaling of DRA groups. Groups are watched on create, startup load, and update (covering enabling DRA on an existing group at runtime). Per evaluation round:

  • Floor: min-parallelism deficits are satisfied immediately in K-thread units, resetting any demand-phase timers so a recovery does not fire through a stale gate.
  • Immediate demand: busy + serviceable PLANNED > effective threads scales out with an exponential ramp (1, 2, 4, 8 instances) clamped to the actual need — Spark ExecutorAllocationManager semantics, including resetting the ramp when the clamp binds or demand clears.
  • Future demand: pending tables while every thread is busy add one probe instance (pending tables are not quantified demand before planning).
  • Demand must persist for scheduler-backlog-timeout before the first scale-out; later rounds are spaced by sustained-backlog-timeout, so a trickle drained between rounds never accumulates. The max-parallelism cap always wins.
  • Failure paths: a transient group-read failure retries instead of being treated as deletion; an enabled group whose queue is momentarily absent (delete/recreate racing the config watcher) is transient too; group deletion cleans all keeper state immediately.

• PendingRegistrations (new): requested-but-unregistered capacity counts toward effective threads so a booting pod is not re-requested every round. Registration is optimizer-driven and heartbeat expiry only starts after it, so entries carry their own boot deadline (3 minutes). A synchronously failed request is dropped and retried; a pod whose requestResource succeeded but whose persistence failed keeps its entry (it will self-register) instead of being re-requested as a duplicate.

• OptimizingQueue.collectDynamicAllocationLoad() (new): snapshots the demand side — busy threads (SCHEDULED and ACKED: a thread is occupied from assignment, not from ack), quota-mode-aware serviceable PLANNED tasks (a proportional quota ≤ 1 counts its whole backlog since scaling raises its ceil(quota × availableCore) limit; an absolute quota > 1 counts only free slots), and PENDING tables. Table runtimes are copied under tableLock via a new SchedulingPolicy.snapshotTableRuntimes() rather than iterating the lock-guarded map from another thread.

• Planning-bound guard: idle registered threads while tables wait as PENDING and nothing is PLANNED means the bottleneck is the serialized planning (optimizer.max-planning-parallelism), not thread capacity — scaling out would only add idle threads. The keeper warns (once per episode, after the condition persists across two evaluations) naming the config to raise, and does not scale.

• Docs: the optimizer group property table now documents all dynamic-allocation.* properties (as promised in the Phase 1 PR for the phase that first exposes runtime behavior), marks the flat min-parallelism deprecated, and recommends executor-parallelism 4–8 for Kubernetes groups.

Known trade-offs

• Proportional-quota convergence: counting a proportional table's whole backlog can over-provision toward E ≈ planned/(1−q) while the table only occupies ceil(q×E) threads; the excess is reclaimed by idle-timeout when scale-down lands (Phase 4), and max-parallelism is the operator's bound until then.
• Manual optimizer operations: a pod created via the dashboard scale-out endpoint is invisible to the boot-window accounting, so the keeper may briefly scale on top of it (bounded; corrects once it registers). Manual operations on DRA groups are documented as not recommended.
• HA failover: boot-window accounting is leader-local, so a new leader can re-request capacity the old leader already requested within the boot window — bounded to one duplicate round and self-correcting on registration.
• Boot deadline: 3 minutes is sized for Kubernetes pods (the AIP's primary target); slow container types may exceed it and briefly double-request, reclaimed by idle-timeout in Phase 4. Making it configurable is left to a follow-up to avoid growing the public config surface here.

How was this patch tested?

• Add some test cases that check the changes thoroughly including negative and positive cases if possible
  • TestComputeScaleUp (15): floor immediacy and cap clamping, backlog/sustained timing gates, exponential ramp including reset-on-clamp-binding, trickle filtering, cold-start ignition, future-demand single-probe semantics, stale-gate reset after floor recovery.
  • TestDynamicAllocationState (17): quota-mode-aware serviceable counting (including fractional absolute quotas), SCHEDULED-occupies-thread semantics, planning-bound predicate.
  • TestPendingRegistrations (6): boot-deadline expiry per entry, registration/failure removal.
  • TestDynamicAllocationConfig (29): executor-parallelism parsing and validation including the unreachable-floor rejection.
  • TestOptimizerScaleKeeper (4, integration): K-unit floor satisfaction (vs. the legacy single deficit-sized instance), boot-window duplicate prevention (the legacy path re-requests the full deficit every round), failure retry, runtime enablement.
  • TestOptimizingQueue#testCollectDynamicAllocationLoad: the PENDING-table cold-start signal and the SCHEDULED transition on a real table.
• Run test locally before making a pull request: full amoro-common and amoro-ams suites pass (the AMS suite takes ~34 min); spotless and checkstyle clean.

Documentation

• Does this pull request introduce a new feature? (yes)
• If yes, how is the feature documented? (docs — the optimizer group property table in docs/admin-guides/managing-optimizers.md; the AIP-5 page will be revised alongside this PR to document dynamic-allocation.executor-parallelism, which replaces the page's incorrect reference to execution-parallel)