j1wonpark opened a new pull request, #4272:
URL: https://github.com/apache/amoro/pull/4272
## 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?
- [x] 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.
- [x] 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`)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
