[AMORO-4295] AIP-5 Phase 3+4: idle-driven scale-down with graceful drain for dynamic allocation groups

[j1wonpark]

Why are the changes needed?

Close #4295.

Third and fourth implementation phase of AIP-5: Dynamic Resource Allocation for Optimizer (#4191), building on the scale-up merged in #4272: per-optimizer idle tracking and idle-driven scale-down with graceful drain for dynamic-allocation-enabled groups. One PR because idle tracking's only consumer is the scale-down decision. Opt-in as before: groups without dynamic-allocation.enabled = true are unaffected.

One deliberate deviation from the AIP page: idle tracking is derived from the per-round load snapshot (per-token in-flight counts) instead of the sketched event-based OptimizerCounters. Several task-reclamation paths offer no service-level release hook (resetStaleTasksForThread, the suspending predication reclaiming timed-out tasks of live optimizers, process close), so an event counter would silently drift — and one leaked increment makes an optimizer permanently non-idle, i.e. a permanent pod leak. Snapshot-derived state is re-derived every round and immune to all of them, consistent with the state-derived position taken in #4272. The AIP page will be revised alongside this PR.

Drain is an efficiency layer only — correctness stays with the existing retry-on-expiry machinery. A draining optimizer stops receiving tasks (pollTask returns null), finishes in-flight work, and is removed; a drain-timeout force removal falls back to the existing orphan recovery.

Brief change log

• OptimizingQueue.collectDynamicAllocationLoad() additionally aggregates per-token in-flight counts; recovered tasks keep their tokens, so the first post-restart snapshot is accurate.
• DynamicAllocationState: observe() tracks last-busy time per token (new tokens are seeded idle, so an instance that never gets a task is still reclaimed); computeScaleDown() applies cooldown → idle filter → floor (registered − draining − candidate ≥ min-parallelism) → longest-idle selection, one per round, and runs only in rounds with no scale-up demand signal (wasDemandActive()). New validation: sustained-backlog-timeout ≤ executor-idle-timeout / 2 (observation resolution equals the keeper cadence).
• Drain plumbing in DefaultOptimizingService: a pending-removal set with deadlines; pollTask blocks draining tokens at entry and re-checks after the internal poll returns (closes the parked long-poll race); executeRemoval follows the dashboard release order, tolerates a missing resource row (the [AMORO-4271] AIP-5 Phase 2: demand-driven scale-up for dynamic allocation groups #4272 boot path), and retries idempotently on transient failures. Only AMS-launched optimizers are candidates.
• OptimizerScaleKeeper round order: drain progress (unconditional) → demand accounting excluding draining instances → idle observation → scale-up → scale-down. Evaluation is time-parameterized behind a @VisibleForTesting hook — our own validation forbids idle timeouts under 30s, so real-time tests are impossible.
• Lifecycle: heartbeat-expiry unregistration clears drain state; unwatching or deleting a group lifts the poll blocks so no draining pod is starved.
• Docs: the scale-down rows in managing-optimizers.md now describe the actual behavior instead of "lands in a later release".

Known trade-offs

• Idle observation is round-granular: a task fitting entirely between two snapshots goes unseen — a frequent-but-bounded error, versus the rare-but-unbounded drift of an event counter.
• All state is leader-local (consistent with [AMORO-4271] AIP-5 Phase 2: demand-driven scale-up for dynamic allocation groups #4272): failover means a delayed scale-down, never a lost task.
• A heartbeat-expiry unregistration racing a removal in the same round can dip the group below its floor for one round; self-correcting, deliberately unguarded (untestable).

How was this patch tested?

• Add some test cases that check the changes thoroughly including negative and positive cases if possible
  • TestComputeScaleDown (10), TestComputeScaleUp (+3), TestDynamicAllocationConfig (+2), TestOptimizingQueue (+2), TestDefaultOptimizingService (+5), TestOptimizerScaleKeeper (+7): idle observation semantics, floor/cooldown selection, the parked long-poll race, executeRemoval failure paths, drain-state cleanup, and idle scale-down end-to-end via injected rounds. Not covered: end-to-end busy-drain and drain-timeout against real running tasks, and a scale-to-zero round trip (each constituent decision is unit-covered).
• Run test locally before making a pull request: full DRA-related suite green (160 tests, Phase 1/2 unmodified); spotless and checkstyle clean.

Documentation

• Does this pull request introduce a new feature? (yes)
• If yes, how is the feature documented? (docs — see the change log; no new configuration. The AIP-5 page will be revised alongside this PR; metrics docs land in the final observability phase)