This is an automated email from the ASF dual-hosted git repository.
zhoujinsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/master by this push:
new 287f4586b [AMORO-4295] AIP-5 Phase 3+4: idle-driven scale-down with
graceful drain for dynamic allocation groups (#4296)
287f4586b is described below
commit 287f4586b2357097e472de2e4868c8f59b8cc412
Author: Jiwon Park <[email protected]>
AuthorDate: Tue Aug 11 20:10:52 2026 +0900
[AMORO-4295] AIP-5 Phase 3+4: idle-driven scale-down with graceful drain
for dynamic allocation groups (#4296)
* [AMORO-4295] Expose per-token in-flight counts in the dynamic allocation
load snapshot
Scale-down needs to know which optimizer instances are busy, not only how
many threads are. Aggregate SCHEDULED/ACKED task counts per optimizer token
in the same scan collectDynamicAllocationLoad() already performs, so one
snapshot supplies both scale directions. Recovered tasks keep their token
across an AMS restart, which makes the very first snapshot after a restart
accurate without any rebuild step.
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4295] Add snapshot-derived idle observation and the scale-down
decision
Track per-token idle time by observation instead of event counters: each
keeper round feeds the task snapshot into observe(), which refreshes busy
timestamps, seeds first-seen tokens as idle (an instance that never
receives a task must still become removable), and prunes unregistered
ones. Event-maintained counters were rejected because three existing
paths reclaim assigned tasks without any service-level decrement hook
(queue-internal stale-ACKED resets, ack/exec-timeout reclaims of live
optimizers, and process close), each silently inflating a counter until
the instance can never look idle again.
computeScaleDown picks at most one candidate per round: the longest-idle
instance past executor-idle-timeout, rate-limited by scale-down-cooldown,
whose removal keeps registered-minus-draining threads at or above
min-parallelism. Draining threads count as already gone so consecutive
removals cannot pass the floor check together and land below the floor
once they all complete.
validate() now also requires sustained-backlog-timeout to be at most half
of executor-idle-timeout: the keeper cadence is the observation
resolution, and sampling slower than that misjudges an instance that
worked between samples as continuously idle.
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4295] Add graceful drain plumbing: poll blocking and optimizer
removal
A draining optimizer's token enters a pending-removal set consulted twice
in pollTask: once on entry, and once after the queue poll returns, because
the entry check cannot stop a thread already parked in the queue's
long-poll — it may fetch a task after the drain began, and that task is
handed back (PLANNED again) instead of being assigned. touch/ack/complete
stay open so in-flight tasks finish normally.
executeRemoval releases the container resource, deletes the persisted row,
and unregisters. A missing resource row is handled by releasing through
the optimizer instance itself (it extends Resource and carries the
container identity): a pod that self-registered after a persist failure
has no row and never will, so treating that as a retryable error would
drain-block it forever while its heartbeat keeps it registered. A failed
container release keeps the drain state and retries on a later round —
Kubernetes deletion is idempotent.
The mock container used by the keeper tests now records released
resources and can simulate release failures.
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4295] Integrate scale-down and drain progress into the scale
keeper round
Each round now runs in a fixed order. Drain progress first and
unconditionally — a drain that completed or hit its deadline converts to
a removal even in rounds that scale up, or a busy drain would linger to
its full timeout while backlog persists. Draining instances are then
accounted as already gone on both sides of the demand math: leaving
their threads in the capacity undercounts demand by up to their thread
count, and leaving their tasks in the load keeps the future-demand
signal (busy >= effective) from ever firing mid-drain. Idle observation
runs every round including scale-up ones, so an instance busy through a
burst does not come out of it looking idle since before the burst began.
Scale-down runs only in rounds with no demand signal at all — including
demand still held back by the backlog gate, which computeScaleUp now
exposes: removing a warm instance right before the gate opens would
free exactly the capacity the next round re-requests. A selected victim
begins a graceful drain and is removed in the same round only if a
snapshot taken after the token entered the pending-removal set shows it
idle; the pre-insert snapshot may miss a task fetched by a long-poll
racing the drain start.
The keeper tests drive rounds with injected times because the validated
minimum executor-idle-timeout (30s) puts real idle waits beyond sane
test durations.
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4295] Clear drain state on unregistration and on dynamic
allocation disable
Two lifecycle paths could strand a token in the pending-removal set. An
optimizer that dies mid-drain is unregistered by heartbeat expiry, and
its token can never be matched again (the replacement pod registers
under a fresh one), so unregisterOptimizer now clears the drain state.
And a group whose dynamic allocation is disabled mid-drain returns to
the legacy floor keeper, where a leftover drain block would starve the
still-running pod forever — unwatching the group re-admits its tokens
to task assignment.
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4295] Fix concurrent optimizer unregistration
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4295] Document scale-down and drain behavior in the optimizer
group property table
Signed-off-by: Jiwon Park <[email protected]>
---------
Signed-off-by: Jiwon Park <[email protected]>
Co-authored-by: ZhouJinsong <[email protected]>
---
.../amoro/server/DefaultOptimizingService.java | 238 +++++++++++++++++++-
.../amoro/server/optimizing/OptimizingQueue.java | 7 +-
.../optimizing/dra/DynamicAllocationConfig.java | 13 ++
.../optimizing/dra/DynamicAllocationState.java | 119 +++++++++-
.../amoro/server/TestDefaultOptimizingService.java | 111 +++++++++
.../amoro/server/TestOptimizerGroupKeeper.java | 19 +-
.../amoro/server/TestOptimizerScaleKeeper.java | 249 ++++++++++++++++++++-
.../server/optimizing/TestOptimizingQueue.java | 65 ++++++
.../optimizing/dra/TestComputeScaleDown.java | 208 +++++++++++++++++
.../server/optimizing/dra/TestComputeScaleUp.java | 26 +++
.../dra/TestDynamicAllocationConfig.java | 19 ++
docs/admin-guides/managing-optimizers.md | 6 +-
12 files changed, 1064 insertions(+), 16 deletions(-)
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java
b/amoro-ams/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java
index 427b6edd5..2e9aa4a28 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java
@@ -116,6 +116,16 @@ public class DefaultOptimizingService extends
StatedPersistentBase
private final Map<String, OptimizingQueue> optimizingQueueByGroup = new
ConcurrentHashMap<>();
private final Map<String, OptimizingQueue> optimizingQueueByToken = new
ConcurrentHashMap<>();
private final Map<String, OptimizerInstance> authOptimizers = new
ConcurrentHashMap<>();
+
+ /**
+ * Tokens of draining optimizers (AIP-5 scale-down): {@link #pollTask}
returns {@code null} for
+ * them, blocking new assignments while in-flight tasks complete normally.
+ */
+ private final Set<String> pendingRemovalTokens =
ConcurrentHashMap.newKeySet();
+
+ /** Force-removal deadline per draining token, the {@code drain-timeout}
safety net. */
+ private final Map<String, Long> drainDeadlines = new ConcurrentHashMap<>();
+
private final OptimizerKeeper optimizerKeeper = new
OptimizerKeeper("optimizer-keeper-thread");
private final OptimizerGroupKeeper optimizerGroupKeeper =
new OptimizerGroupKeeper("optimizer-group-keeper-thread");
@@ -242,9 +252,18 @@ public class DefaultOptimizingService extends
StatedPersistentBase
doAs(OptimizerMapper.class, mapper -> mapper.deleteOptimizer(token));
OptimizingQueue optimizingQueue = optimizingQueueByToken.remove(token);
OptimizerInstance optimizer = authOptimizers.remove(token);
- if (optimizingQueue != null) {
- optimizingQueue.removeOptimizer(optimizer);
+ if (optimizer != null) {
+ if (optimizingQueue == null) {
+ optimizingQueue = optimizingQueueByGroup.get(optimizer.getGroupName());
+ }
+ if (optimizingQueue != null) {
+ optimizingQueue.removeOptimizer(optimizer);
+ }
}
+ // An optimizer that dies mid-drain is unregistered here by heartbeat
expiry; its token can
+ // never be matched again, so leftover drain state would sit in the
pending-removal set
+ // forever (its replacement pod registers under a fresh token).
+ cancelDrain(token);
}
@Override
@@ -269,10 +288,15 @@ public class DefaultOptimizingService extends
StatedPersistentBase
@Override
public OptimizingTask pollTask(String authToken, int threadId) {
+ if (pendingRemovalTokens.contains(authToken)) {
+ return null;
+ }
LOG.debug("Optimizer {} (threadId {}) try polling task", authToken,
threadId);
OptimizerThread optimizerThread =
getAuthenticatedOptimizer(authToken).getThread(threadId);
OptimizingQueue queue = getQueueByToken(authToken);
- TaskRuntime<?> task = queue.pollTask(optimizerThread, pollingTimeout,
breakQuotaLimit);
+ TaskRuntime<?> task =
+ guardDrainedPoll(
+ authToken, queue.pollTask(optimizerThread, pollingTimeout,
breakQuotaLimit));
if (task != null) {
LOG.info("OptimizerThread {} polled task {}", optimizerThread,
task.getTaskId());
return task.extractProtocolTask();
@@ -280,6 +304,102 @@ public class DefaultOptimizingService extends
StatedPersistentBase
return null;
}
+ /**
+ * Close the long-poll race on drain start: the entry check above cannot
stop a thread already
+ * parked inside the queue's poll, which may fetch a task after its token
entered the
+ * pending-removal set. Hand such a task back instead of assigning it to a
draining optimizer.
+ */
+ @VisibleForTesting
+ TaskRuntime<?> guardDrainedPoll(String authToken, TaskRuntime<?> task) {
+ if (task == null || !pendingRemovalTokens.contains(authToken)) {
+ return task;
+ }
+ OptimizingQueue queue = optimizingQueueByToken.get(authToken);
+ if (queue != null) {
+ try {
+ queue.retryTask(task);
+ } catch (Exception e) {
+ // The existing suspending-task safety net will still reclaim it after
the removal.
+ LOG.warn(
+ "Failed to hand back task {} from draining optimizer {}",
+ task.getTaskId(),
+ authToken,
+ e);
+ }
+ }
+ return null;
+ }
+
+ /** Block new task assignments to the token; in-flight tasks keep completing
normally. */
+ void beginGracefulDrain(String token, long deadlineMs) {
+ drainDeadlines.put(token, deadlineMs);
+ pendingRemovalTokens.add(token);
+ LOG.info("Optimizer {} begins graceful drain", token);
+ }
+
+ /** Re-admit the token to task assignment, e.g. when dynamic allocation is
disabled mid-drain. */
+ void cancelDrain(String token) {
+ pendingRemovalTokens.remove(token);
+ drainDeadlines.remove(token);
+ }
+
+ @VisibleForTesting
+ boolean isDraining(String token) {
+ return pendingRemovalTokens.contains(token);
+ }
+
+ /**
+ * Run one dynamic-allocation round for the group at an injected time. The
production cadence is
+ * driven by the scale keeper's delay queue with the wall clock; tests
inject times because the
+ * validated minimum {@code executor-idle-timeout} (30s) puts real idle
waits beyond sane test
+ * durations.
+ */
+ @VisibleForTesting
+ void evaluateDynamicAllocation(String groupName, long nowMs) {
+ ResourceGroup resourceGroup = optimizerManager.getResourceGroup(groupName);
+ OptimizingQueue queue = optimizingQueueByGroup.get(groupName);
+ optimizerScaleKeeper.scaleIfNeeded(
+ resourceGroup, queue, DynamicAllocationConfig.parse(resourceGroup),
nowMs);
+ }
+
+ /**
+ * Remove a drained optimizer: release the container resource, delete the
persisted resource row,
+ * and unregister. A missing resource row (the pod self-registered after a
persist failure, or a
+ * manual release raced the row away) is not an error — the instance itself
carries the
+ * container-side identity, so release through it and only skip the row
delete; treating this as a
+ * retryable failure would loop forever on a pod whose row can never
reappear. A container release
+ * failure keeps the drain state so a later round retries the idempotent
deletion.
+ */
+ void executeRemoval(String token) {
+ OptimizerInstance optimizer = authOptimizers.get(token);
+ if (optimizer == null || optimizer.getResourceId() == null) {
+ // Already unregistered, or externally launched: nothing for AMS to
release.
+ cancelDrain(token);
+ return;
+ }
+ try {
+ Resource resource =
optimizerManager.getResource(optimizer.getResourceId());
+ if (resource != null) {
+ resource.getProperties().putAll(optimizer.getProperties());
+ ((AbstractOptimizerContainer)
Containers.get(resource.getContainerName()))
+ .releaseResource(resource);
+ optimizerManager.deleteResource(optimizer.getResourceId());
+ } else {
+ ((AbstractOptimizerContainer)
Containers.get(optimizer.getContainerName()))
+ .releaseResource(optimizer);
+ }
+ } catch (Throwable t) {
+ LOG.warn(
+ "Failed to release optimizer {} (resource {}), will retry",
+ token,
+ optimizer.getResourceId(),
+ t);
+ return;
+ }
+ unregisterOptimizer(token);
+ LOG.info("Optimizer {} (resource {}) removed by scale-down", token,
optimizer.getResourceId());
+ }
+
@Override
public void ackTask(String authToken, int threadId, OptimizingTaskId taskId)
{
LOG.info("Ack task {} by optimizer {} (threadId {})", taskId, authToken,
threadId);
@@ -1046,6 +1166,12 @@ public class DefaultOptimizingService extends
StatedPersistentBase
watchedGroups.remove(groupName);
scaleStates.remove(groupName);
planningBoundStreaks.remove(groupName);
+ // A drain block left behind would starve the group's pods forever once
the legacy floor
+ // keeper resumes duty for the disabled group: re-admit them to task
assignment.
+ authOptimizers.values().stream()
+ .filter(optimizer -> groupName.equals(optimizer.getGroupName()))
+ .map(OptimizerInstance::getToken)
+ .forEach(DefaultOptimizingService.this::cancelDrain);
// pendingRegistrations is deliberately kept: a pod requested before a
disable survives its
// boot window, so re-enabling within it does not re-request the same
capacity. Entries
// self-prune past their deadline.
@@ -1096,7 +1222,7 @@ public class DefaultOptimizingService extends
StatedPersistentBase
}
DynamicAllocationConfig config =
DynamicAllocationConfig.parse(resourceGroup);
try {
- scaleIfNeeded(resourceGroup, queue, config);
+ scaleIfNeeded(resourceGroup, queue, config,
System.currentTimeMillis());
} catch (Throwable t) {
LOG.error("Dynamic allocation scale evaluation failed for group {}",
task.groupName, t);
} finally {
@@ -1111,6 +1237,81 @@ public class DefaultOptimizingService extends
StatedPersistentBase
return pending == null ? 0 :
pending.pendingThreads(System.currentTimeMillis());
}
+ /**
+ * Advance this group's drains: an entry whose in-flight count reached
zero, or whose {@code
+ * drain-timeout} deadline passed, executes its removal now (a
force-removed instance's orphaned
+ * tasks are reclaimed by the existing suspending-task safety net).
Returns the thread and
+ * busy-task counts of instances still draining afterwards — a failed
release keeps its instance
+ * in both, since it remains registered.
+ */
+ private int[] processDrainProgress(
+ String groupName, DynamicAllocationState.GroupLoad load, long now) {
+ int drainingThreads = 0;
+ int drainingBusy = 0;
+ for (String token : pendingRemovalTokens) {
+ OptimizerInstance optimizer = authOptimizers.get(token);
+ if (optimizer == null) {
+ // Unregistered mid-drain (e.g. its heartbeat expired): nothing left
to remove.
+ cancelDrain(token);
+ continue;
+ }
+ if (!groupName.equals(optimizer.getGroupName())) {
+ continue;
+ }
+ int inFlight = load.getInFlightByToken().getOrDefault(token, 0);
+ Long deadline = drainDeadlines.get(token);
+ if (inFlight == 0 || (deadline != null && now >= deadline)) {
+ executeRemoval(token);
+ if (!authOptimizers.containsKey(token)) {
+ continue;
+ }
+ }
+ drainingThreads += optimizer.getThreadCount();
+ drainingBusy += inFlight;
+ }
+ return new int[] {drainingThreads, drainingBusy};
+ }
+
+ private Set<String> registeredTokens(String groupName) {
+ return authOptimizers.values().stream()
+ .filter(optimizer -> groupName.equals(optimizer.getGroupName()))
+ .map(OptimizerInstance::getToken)
+ .collect(Collectors.toSet());
+ }
+
+ private void evaluateScaleDown(
+ String groupName,
+ OptimizingQueue queue,
+ DynamicAllocationState state,
+ DynamicAllocationConfig config,
+ int registeredThreads,
+ int drainingThreads,
+ long now) {
+ List<DynamicAllocationState.RemovalCandidate> candidates =
+ authOptimizers.values().stream()
+ // Externally-registered optimizers (no resourceId) are not
AMS's to remove.
+ .filter(optimizer -> groupName.equals(optimizer.getGroupName()))
+ .filter(optimizer -> optimizer.getResourceId() != null)
+ .filter(optimizer ->
!pendingRemovalTokens.contains(optimizer.getToken()))
+ .map(
+ optimizer ->
+ new DynamicAllocationState.RemovalCandidate(
+ optimizer.getToken(), optimizer.getThreadCount()))
+ .collect(Collectors.toList());
+ String victim =
+ state.computeScaleDown(candidates, registeredThreads,
drainingThreads, config, now);
+ if (victim == null) {
+ return;
+ }
+ beginGracefulDrain(victim, now + config.getDrainTimeout().toMillis());
+ // Only a snapshot taken after the token entered the pending-removal set
can prove idleness:
+ // the pre-insert one may miss a task fetched by a long-poll racing the
drain start.
+ DynamicAllocationState.GroupLoad fresh =
queue.collectDynamicAllocationLoad();
+ if (fresh.getInFlightByToken().getOrDefault(victim, 0) == 0) {
+ executeRemoval(victim);
+ }
+ }
+
private void recheckAfterUnwatch(String groupName) {
try {
ResourceGroup fresh = optimizerManager.getResourceGroup(groupName);
@@ -1157,27 +1358,46 @@ public class DefaultOptimizingService extends
StatedPersistentBase
}
private void scaleIfNeeded(
- ResourceGroup resourceGroup, OptimizingQueue queue,
DynamicAllocationConfig config) {
+ ResourceGroup resourceGroup,
+ OptimizingQueue queue,
+ DynamicAllocationConfig config,
+ long now) {
String groupName = resourceGroup.getName();
- long now = System.currentTimeMillis();
PendingRegistrations pending =
pendingRegistrations.computeIfAbsent(
groupName, name -> new PendingRegistrations(BOOT_TIMEOUT_MS));
DynamicAllocationState state =
scaleStates.computeIfAbsent(groupName, name -> new
DynamicAllocationState());
- int registeredThreads = getTotalQuota(groupName);
- int effectiveThreads = registeredThreads + pending.pendingThreads(now);
DynamicAllocationState.GroupLoad load =
queue.collectDynamicAllocationLoad();
+ // Drain progress runs before anything else and unconditionally: a
completed or expired
+ // drain must convert to a removal even in rounds that scale up, or a
busy drain would
+ // linger to its full timeout while backlog persists.
+ int[] draining = processDrainProgress(groupName, load, now);
+ int drainingThreads = draining[0];
+ int drainingBusy = draining[1];
+ int registeredThreads = getTotalQuota(groupName);
+ // A draining instance takes no new work, so it is accounted as already
gone on both sides:
+ // leaving its threads in the capacity undercounts demand by up to their
count, and leaving
+ // its tasks in the load keeps future demand (busy >= effective) from
ever firing mid-drain.
+ int effectiveThreads = registeredThreads - drainingThreads +
pending.pendingThreads(now);
+ int busyThreads = load.getBusyThreads() - drainingBusy;
+ // Observed every round, including scale-up ones: an instance busy
through a burst must not
+ // come out of it looking idle since before the burst began.
+ state.observe(registeredTokens(groupName), load.getInFlightByToken(),
now);
warnOnPlanningBoundTransition(groupName, registeredThreads, load);
int addInstances =
state.computeScaleUp(
effectiveThreads,
- load.getBusyThreads(),
+ busyThreads,
load.getServiceablePlanned(),
load.getPendingTables(),
config,
now);
if (addInstances <= 0) {
+ if (!state.wasDemandActive()) {
+ evaluateScaleDown(
+ groupName, queue, state, config, registeredThreads,
drainingThreads, now);
+ }
return;
}
int threadsPerInstance = config.getExecutorParallelism();
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
index bbbd6b288..178bebaa3 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
@@ -428,11 +428,13 @@ public class OptimizingQueue extends PersistentBase {
public DynamicAllocationState.GroupLoad collectDynamicAllocationLoad() {
Map<Long, Integer> plannedByTable = Maps.newHashMap();
Map<Long, Integer> occupiedByTable = Maps.newHashMap();
+ Map<String, Integer> inFlightByToken = Maps.newHashMap();
int busyThreads = 0;
for (TaskRuntime<?> task : collectTasks()) {
if (DynamicAllocationState.occupiesThread(task.getStatus())) {
busyThreads++;
occupiedByTable.merge(task.getTableId(), 1, Integer::sum);
+ inFlightByToken.merge(task.getToken(), 1, Integer::sum);
} else if (task.getStatus() == Status.PLANNED) {
plannedByTable.merge(task.getTableId(), 1, Integer::sum);
}
@@ -457,7 +459,10 @@ public class OptimizingQueue extends PersistentBase {
targetQuotaByTable.getOrDefault(tableId, 1.0),
occupiedByTable.getOrDefault(tableId, 0))));
return new DynamicAllocationState.GroupLoad(
- busyThreads, DynamicAllocationState.serviceablePlannedCount(demands),
pendingTables);
+ busyThreads,
+ DynamicAllocationState.serviceablePlannedCount(demands),
+ pendingTables,
+ inFlightByToken);
}
public void retryTask(TaskRuntime<?> taskRuntime) {
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationConfig.java
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationConfig.java
index 15c9de841..cfa0e2f1d 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationConfig.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationConfig.java
@@ -338,6 +338,19 @@ public class DynamicAllocationConfig {
OptimizerProperties.DYNAMIC_ALLOCATION_SUSTAINED_BACKLOG_TIMEOUT,
sustainedBacklogTimeout);
requirePositive(OptimizerProperties.DYNAMIC_ALLOCATION_SCALE_DOWN_COOLDOWN,
scaleDownCooldown);
requirePositive(OptimizerProperties.DYNAMIC_ALLOCATION_DRAIN_TIMEOUT,
drainTimeout);
+ // The scale keeper evaluates each group at sustained-backlog-timeout
cadence, which is also
+ // the idle observation resolution: sampling slower than half the idle
timeout lets an
+ // instance that worked between samples be misjudged as continuously idle
and drained.
+ if (sustainedBacklogTimeout.toMillis() * 2 >
executorIdleTimeout.toMillis()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Resource group:%s '%s'(%s) must be <= half of '%s'(%s).",
+ groupName,
+ OptimizerProperties.DYNAMIC_ALLOCATION_SUSTAINED_BACKLOG_TIMEOUT,
+ sustainedBacklogTimeout,
+ OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_IDLE_TIMEOUT,
+ executorIdleTimeout));
+ }
}
private void requirePositive(String property, Duration value) {
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java
index 9ac84334a..e0cd45057 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java
@@ -21,6 +21,10 @@ package org.apache.amoro.server.optimizing.dra;
import org.apache.amoro.server.optimizing.TaskRuntime;
import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
/**
* Per-group scale-up decision state for dynamic allocation (AIP-5): the
backlog timer, the
@@ -39,6 +43,19 @@ public final class DynamicAllocationState {
/** Instances to add in the next immediate-demand round (1, 2, 4, 8 ...). */
private int rampInstances = 1;
+ /**
+ * Last time each registered token was observed with in-flight tasks; seeded
with the first
+ * observation time, so a fresh instance is idle from creation and a
permanently unused one still
+ * becomes a removal candidate.
+ */
+ private final Map<String, Long> lastBusyMs = new HashMap<>();
+
+ /** Time of the last scale-down selection; {@code -1} before the first one.
*/
+ private long lastScaleDownMs = -1;
+
+ /** Whether the last {@link #computeScaleUp} evaluation saw any demand (or a
floor deficit). */
+ private boolean lastEvalDemandActive = false;
+
/**
* Decide how many executor-parallelism-thread optimizer instances to create
in this round.
*
@@ -68,6 +85,7 @@ public final class DynamicAllocationState {
// A floor deficit (optimizers died or the group is new) invalidates any
demand-phase
// state: after recovery, demand must re-prove backlog persistence
instead of firing
// through a stale gate with a stale ramp.
+ lastEvalDemandActive = true;
backlogSinceMs = -1;
nextAllowedAddMs = -1;
rampInstances = 1;
@@ -77,6 +95,7 @@ public final class DynamicAllocationState {
int actionableNeed = Math.max(busyThreads + serviceablePlanned -
effectiveThreads, 0);
boolean futureDemand = pendingTables > 0 && busyThreads >=
effectiveThreads;
+ lastEvalDemandActive = actionableNeed > 0 || futureDemand;
if (actionableNeed <= 0 && !futureDemand) {
backlogSinceMs = -1;
nextAllowedAddMs = -1;
@@ -117,6 +136,93 @@ public final class DynamicAllocationState {
return add;
}
+ /**
+ * Whether the last {@link #computeScaleUp} evaluation saw demand —
including a floor deficit and
+ * demand still held back by the backlog gate. Scale-down must not run in
such a round: it would
+ * remove exactly the warm capacity the next round re-requests.
+ */
+ public boolean wasDemandActive() {
+ return lastEvalDemandActive;
+ }
+
+ /**
+ * Update per-token idle observations from this round's snapshot. A token
with in-flight tasks has
+ * its busy timestamp refreshed; a token seen for the first time is seeded
with {@code nowMs}
+ * (idle from first sight, see {@link #lastBusyMs}); tokens no longer
registered are pruned, so a
+ * re-registered identity re-earns its idle time instead of inheriting a
stale timestamp.
+ */
+ public void observe(
+ Set<String> registeredTokens, Map<String, Integer> inFlightByToken, long
nowMs) {
+ lastBusyMs.keySet().retainAll(registeredTokens);
+ for (String token : registeredTokens) {
+ if (inFlightByToken.getOrDefault(token, 0) > 0 ||
!lastBusyMs.containsKey(token)) {
+ lastBusyMs.put(token, nowMs);
+ }
+ }
+ }
+
+ /**
+ * Pick at most one optimizer to drain this round, or {@code null}. The
caller passes only
+ * eligible candidates (registered, AMS-launched, not already draining).
Selection: skip inside
+ * the {@code scale-down-cooldown} window; among candidates idle for at
least {@code
+ * executor-idle-timeout} whose removal keeps {@code registeredThreads -
drainingThreads} at or
+ * above the floor, pick the longest-idle one. Draining threads are counted
as already gone —
+ * still-registered draining instances must not let consecutive removals
pass the floor check and
+ * land the group below {@code min-parallelism} once they all complete.
+ */
+ public String computeScaleDown(
+ List<RemovalCandidate> candidates,
+ int registeredThreads,
+ int drainingThreads,
+ DynamicAllocationConfig config,
+ long nowMs) {
+ if (lastScaleDownMs >= 0
+ && nowMs - lastScaleDownMs < config.getScaleDownCooldown().toMillis())
{
+ return null;
+ }
+ long idleTimeoutMs = config.getExecutorIdleTimeout().toMillis();
+ int floorBase = registeredThreads - drainingThreads;
+ RemovalCandidate selected = null;
+ long selectedBusyMs = Long.MAX_VALUE;
+ for (RemovalCandidate candidate : candidates) {
+ Long busy = lastBusyMs.get(candidate.getToken());
+ if (busy == null || nowMs - busy < idleTimeoutMs) {
+ continue;
+ }
+ if (floorBase - candidate.getThreadCount() < config.getMinParallelism())
{
+ continue;
+ }
+ if (busy < selectedBusyMs) {
+ selectedBusyMs = busy;
+ selected = candidate;
+ }
+ }
+ if (selected == null) {
+ return null;
+ }
+ lastScaleDownMs = nowMs;
+ return selected.getToken();
+ }
+
+ /** A removal candidate: a registered, AMS-launched, not-yet-draining
optimizer instance. */
+ public static class RemovalCandidate {
+ private final String token;
+ private final int threadCount;
+
+ public RemovalCandidate(String token, int threadCount) {
+ this.token = token;
+ this.threadCount = threadCount;
+ }
+
+ public String getToken() {
+ return token;
+ }
+
+ public int getThreadCount() {
+ return threadCount;
+ }
+ }
+
private static int ceilDiv(int value, int divisor) {
return (value + divisor - 1) / divisor;
}
@@ -126,11 +232,17 @@ public final class DynamicAllocationState {
private final int busyThreads;
private final int serviceablePlanned;
private final int pendingTables;
+ private final Map<String, Integer> inFlightByToken;
- public GroupLoad(int busyThreads, int serviceablePlanned, int
pendingTables) {
+ public GroupLoad(
+ int busyThreads,
+ int serviceablePlanned,
+ int pendingTables,
+ Map<String, Integer> inFlightByToken) {
this.busyThreads = busyThreads;
this.serviceablePlanned = serviceablePlanned;
this.pendingTables = pendingTables;
+ this.inFlightByToken = inFlightByToken;
}
public int getBusyThreads() {
@@ -144,6 +256,11 @@ public final class DynamicAllocationState {
public int getPendingTables() {
return pendingTables;
}
+
+ /** SCHEDULED/ACKED task count per optimizer token, the per-instance side
of the snapshot. */
+ public Map<String, Integer> getInFlightByToken() {
+ return inFlightByToken;
+ }
}
/** Per-table demand snapshot consumed by {@link
#serviceablePlannedCount(Collection)}. */
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/TestDefaultOptimizingService.java
b/amoro-ams/src/test/java/org/apache/amoro/server/TestDefaultOptimizingService.java
index ff2c94a42..5c9d52554 100644
---
a/amoro-ams/src/test/java/org/apache/amoro/server/TestDefaultOptimizingService.java
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/TestDefaultOptimizingService.java
@@ -18,6 +18,8 @@
package org.apache.amoro.server;
+import static
org.apache.amoro.server.optimizing.OptimizerGroupMetrics.OPTIMIZER_GROUP_OPTIMIZER_INSTANCES;
+import static
org.apache.amoro.server.optimizing.OptimizerGroupMetrics.OPTIMIZER_GROUP_THREADS;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
@@ -38,10 +40,15 @@ import org.apache.amoro.config.TableConfiguration;
import org.apache.amoro.exception.PluginRetryAuthException;
import org.apache.amoro.exception.TaskRuntimeException;
import org.apache.amoro.io.MixedDataTestHelpers;
+import org.apache.amoro.metrics.Gauge;
+import org.apache.amoro.metrics.MetricKey;
+import org.apache.amoro.metrics.MetricRegistry;
import org.apache.amoro.optimizing.RewriteFilesOutput;
import org.apache.amoro.optimizing.TableOptimizing;
import org.apache.amoro.process.ProcessStatus;
import org.apache.amoro.resource.ResourceGroup;
+import org.apache.amoro.server.manager.MetricManager;
+import org.apache.amoro.server.optimizing.OptimizingQueue;
import org.apache.amoro.server.optimizing.OptimizingStatus;
import org.apache.amoro.server.optimizing.TaskRuntime;
import org.apache.amoro.server.persistence.SqlSessionFactoryProvider;
@@ -70,6 +77,7 @@ import org.junit.jupiter.api.Assertions;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import java.lang.reflect.Field;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
@@ -180,6 +188,35 @@ public class TestDefaultOptimizingService extends
AMSTableTestBase {
assertTaskCompleted(taskRuntime);
}
+ @Test
+ public void testPollTaskBlockedWhileDraining() {
+ // A draining optimizer receives no new assignments even though a task is
available; in-flight
+ // completion paths (touch/ack/complete) are deliberately not blocked.
+ optimizingService().beginGracefulDrain(token, Long.MAX_VALUE);
+ Assertions.assertNull(optimizingService().pollTask(token, THREAD_ID));
+
+ optimizingService().cancelDrain(token);
+ Assertions.assertNotNull(optimizingService().pollTask(token, THREAD_ID));
+ }
+
+ @Test
+ public void testDrainStartedDuringPollHandsTaskBack() {
+ OptimizingTask polled = optimizingService().pollTask(token, THREAD_ID);
+ Assertions.assertNotNull(polled);
+ TaskRuntime<?> taskRuntime =
+
optimizingService().listTasks(defaultResourceGroup().getName()).stream()
+ .filter(t -> t.getStatus() == TaskRuntime.Status.SCHEDULED)
+ .findFirst()
+ .orElse(null);
+ Assertions.assertNotNull(taskRuntime);
+
+ // The drain begins while a long-poll is parked inside the queue: the
entry check has already
+ // passed, so the post-poll guard must hand the fetched task back instead
of assigning it.
+ optimizingService().beginGracefulDrain(token, Long.MAX_VALUE);
+ Assertions.assertNull(optimizingService().guardDrainedPoll(token,
taskRuntime));
+ Assertions.assertEquals(TaskRuntime.Status.PLANNED,
taskRuntime.getStatus());
+ }
+
@Test
public void testPollTaskTwice() {
// 1.poll task
@@ -265,6 +302,73 @@ public class TestDefaultOptimizingService extends
AMSTableTestBase {
Assertions.assertTrue(optimizerAfterTouched.getTouchTime() > oldTouchTime);
}
+ @Test
+ public void testHeartbeatExpiryClearsDrainState() throws
InterruptedException {
+ // An optimizer that dies mid-drain is unregistered by heartbeat expiry, a
path that must
+ // clear the drain state too: the token can never be matched again, so a
leftover entry would
+ // sit in the pending-removal set forever.
+ rebootWithHeartbeatTimeout(EXPIRATION_TEST_HEARTBEAT_TIMEOUT);
+ String drainingToken = token;
+ optimizingService().beginGracefulDrain(drainingToken, Long.MAX_VALUE);
+ toucher.stop();
+ toucher = null;
+ waitForOptimizerExpiration(drainingToken, ASYNC_WAIT_TIMEOUT_MS);
+ Assertions.assertThrows(
+ PluginRetryAuthException.class, () ->
optimizingService().touch(drainingToken));
+ Assertions.assertFalse(
+ optimizingService().isDraining(drainingToken),
+ "unregistration must clear the drain state of a dead optimizer");
+ }
+
+ @Test
+ public void testUnregisterDoesNotFailWhenAuthenticationAlreadyRemoved()
throws Exception {
+ toucher.stop();
+ toucher = null;
+ OptimizerInstance optimizer = optimizerManager().listOptimizers().get(0);
+ OptimizingQueue queue = (OptimizingQueue)
optimizerState("optimizingQueueByToken").get(token);
+ // Simulate another unregister call having already claimed the
authentication entry.
+ optimizerState("authOptimizers").remove(token);
+
+ try {
+ Assertions.assertDoesNotThrow(
+ () ->
+ optimizingService()
+ .deleteOptimizer(optimizer.getGroupName(),
optimizer.getResourceId()));
+ } finally {
+ queue.removeOptimizer(optimizer);
+ }
+ }
+
+ @Test
+ public void testUnregisterCleansMetricsWhenTokenQueueAlreadyRemoved() throws
Exception {
+ toucher.stop();
+ toucher = null;
+ OptimizerInstance optimizer = optimizerManager().listOptimizers().get(0);
+ // Simulate another unregister call having already claimed the
token-to-queue entry.
+ OptimizingQueue queue =
+ (OptimizingQueue)
optimizerState("optimizingQueueByToken").remove(token);
+ Map<String, String> tagValues = Maps.newHashMap();
+ tagValues.put("group", optimizer.getGroupName());
+ MetricRegistry registry = MetricManager.getInstance().getGlobalRegistry();
+ Gauge<Integer> optimizerCountGauge =
+ (Gauge<Integer>)
+ registry
+ .getMetrics()
+ .get(new MetricKey(OPTIMIZER_GROUP_OPTIMIZER_INSTANCES,
tagValues));
+ Gauge<Long> optimizerThreadsGauge =
+ (Gauge<Long>) registry.getMetrics().get(new
MetricKey(OPTIMIZER_GROUP_THREADS, tagValues));
+
+ Assertions.assertEquals(1, optimizerCountGauge.getValue());
+ Assertions.assertEquals(1L, optimizerThreadsGauge.getValue());
+ try {
+ optimizingService().deleteOptimizer(optimizer.getGroupName(),
optimizer.getResourceId());
+ Assertions.assertEquals(0, optimizerCountGauge.getValue());
+ Assertions.assertEquals(0L, optimizerThreadsGauge.getValue());
+ } finally {
+ queue.removeOptimizer(optimizer);
+ }
+ }
+
@Test
public void testTouchTimeout() throws InterruptedException {
rebootWithHeartbeatTimeout(EXPIRATION_TEST_HEARTBEAT_TIMEOUT);
@@ -728,6 +832,13 @@ public class TestDefaultOptimizingService extends
AMSTableTestBase {
return registerInfo;
}
+ @SuppressWarnings("unchecked")
+ private Map<String, ?> optimizerState(String fieldName) throws Exception {
+ Field field = DefaultOptimizingService.class.getDeclaredField(fieldName);
+ field.setAccessible(true);
+ return (Map<String, ?>) field.get(optimizingService());
+ }
+
private OptimizingTaskResult buildOptimizingTaskResult(OptimizingTaskId
taskId) {
TableOptimizing.OptimizingOutput output = new RewriteFilesOutput(null,
null, null);
OptimizingTaskResult optimizingTaskResult = new
OptimizingTaskResult(taskId, THREAD_ID);
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerGroupKeeper.java
b/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerGroupKeeper.java
index 20d4504cc..6e0591d18 100644
---
a/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerGroupKeeper.java
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerGroupKeeper.java
@@ -45,7 +45,9 @@ import org.junit.jupiter.api.Assertions;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import java.util.List;
import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
@@ -403,6 +405,8 @@ public class TestOptimizerGroupKeeper extends
AMSTableTestBase {
private final AtomicInteger scaleOutCallCount;
private final Function<OptimizerRegisterInfo, String> optimizerRegistrar;
private final Supplier<String> targetGroupNameSupplier;
+ private final AtomicBoolean releaseAvailable = new AtomicBoolean(true);
+ private final List<Resource> releasedResources = new
CopyOnWriteArrayList<>();
public MockOptimizerContainer(
AtomicBoolean resourceAvailable,
@@ -444,6 +448,19 @@ public class TestOptimizerGroupKeeper extends
AMSTableTestBase {
}
@Override
- public void releaseResource(Resource resource) {}
+ public void releaseResource(Resource resource) {
+ if (!releaseAvailable.get()) {
+ throw new RuntimeException("release failed");
+ }
+ releasedResources.add(resource);
+ }
+
+ public void setReleaseAvailable(boolean available) {
+ releaseAvailable.set(available);
+ }
+
+ public List<Resource> getReleasedResources() {
+ return releasedResources;
+ }
}
}
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerScaleKeeper.java
b/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerScaleKeeper.java
index 8a57acb6c..ab913232c 100644
---
a/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerScaleKeeper.java
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerScaleKeeper.java
@@ -64,6 +64,7 @@ public class TestOptimizerScaleKeeper extends
AMSTableTestBase {
private volatile Function<org.apache.amoro.api.OptimizerRegisterInfo,
String> optimizerRegistrar;
private static boolean originIsInitialized = false;
private String currentGroupName;
+ private TestOptimizerGroupKeeper.MockOptimizerContainer mockContainer;
public TestOptimizerScaleKeeper(
CatalogTestHelper catalogTestHelper, TableTestHelper tableTestHelper) {
@@ -120,7 +121,7 @@ public class TestOptimizerScaleKeeper extends
AMSTableTestBase {
}
private void setupMockContainer(Supplier<String> targetGroupNameSupplier)
throws Exception {
- TestOptimizerGroupKeeper.MockOptimizerContainer mockContainer =
+ mockContainer =
new TestOptimizerGroupKeeper.MockOptimizerContainer(
resourceAvailable,
scaleOutCallCount,
@@ -283,4 +284,250 @@ public class TestOptimizerScaleKeeper extends
AMSTableTestBase {
Assertions.assertEquals(
2, optimizers.size(), "runtime-enabled DRA group should reach its
floor in K units");
}
+
+ private OptimizerInstance awaitSingleOptimizer(String groupName) throws
InterruptedException {
+ Thread.sleep(500);
+ List<OptimizerInstance> optimizers =
optimizerManager().listOptimizers(groupName);
+ Assertions.assertEquals(1, optimizers.size(), "floor of 1 should register
one optimizer");
+ return optimizers.get(0);
+ }
+
+ /** Removal releases the container resource, deletes the persisted row, and
unregisters. */
+ @Test
+ public void testExecuteRemovalReleasesResourceAndUnregisters() throws
InterruptedException {
+ resourceAvailable.set(true);
+ scaleOutCallCount.set(0);
+ ResourceGroup group = buildDraResourceGroup(TEST_GROUP_NAME + "-5", 1, 1);
+ optimizerManager().createResourceGroup(group);
+ optimizingService().createResourceGroup(group);
+ OptimizerInstance optimizer = awaitSingleOptimizer(group.getName());
+
+ // Keep the keeper from instantly re-filling the floor while we assert
emptiness.
+ resourceAvailable.set(false);
+ optimizingService().executeRemoval(optimizer.getToken());
+
+
Assertions.assertTrue(optimizerManager().listOptimizers(group.getName()).isEmpty());
+
Assertions.assertNull(optimizerManager().getResource(optimizer.getResourceId()));
+ Assertions.assertTrue(
+ mockContainer.getReleasedResources().stream()
+ .anyMatch(r ->
optimizer.getResourceId().equals(r.getResourceId())),
+ "the container resource must be released");
+ }
+
+ /**
+ * A registered optimizer whose resource row is missing (the pod
self-registered after a persist
+ * failure, or a manual release raced the row away) must still be removable:
the instance itself
+ * carries the container-side identity, so release through it and only skip
the row delete.
+ */
+ @Test
+ public void testExecuteRemovalFallsBackWhenResourceRowMissing() throws
InterruptedException {
+ resourceAvailable.set(true);
+ scaleOutCallCount.set(0);
+ ResourceGroup group = buildDraResourceGroup(TEST_GROUP_NAME + "-6", 1, 1);
+ optimizerManager().createResourceGroup(group);
+ optimizingService().createResourceGroup(group);
+ OptimizerInstance optimizer = awaitSingleOptimizer(group.getName());
+
+ resourceAvailable.set(false);
+ optimizerManager().deleteResource(optimizer.getResourceId());
+ optimizingService().executeRemoval(optimizer.getToken());
+
+ Assertions.assertTrue(
+ optimizerManager().listOptimizers(group.getName()).isEmpty(),
+ "a row-less optimizer must not become an unremovable zombie");
+ Assertions.assertTrue(
+ mockContainer.getReleasedResources().stream()
+ .anyMatch(r ->
optimizer.getResourceId().equals(r.getResourceId())),
+ "the container side must still be released via the instance");
+ }
+
+ /**
+ * A transient container release failure keeps the drain state so a later
round retries the
+ * idempotent deletion; the optimizer must not be unregistered while its pod
may still exist.
+ */
+ @Test
+ public void testExecuteRemovalKeepsDrainStateOnReleaseFailure() throws
InterruptedException {
+ resourceAvailable.set(true);
+ scaleOutCallCount.set(0);
+ ResourceGroup group = buildDraResourceGroup(TEST_GROUP_NAME + "-7", 1, 1);
+ optimizerManager().createResourceGroup(group);
+ optimizingService().createResourceGroup(group);
+ OptimizerInstance optimizer = awaitSingleOptimizer(group.getName());
+
+ resourceAvailable.set(false);
+ optimizingService().beginGracefulDrain(optimizer.getToken(),
Long.MAX_VALUE);
+ mockContainer.setReleaseAvailable(false);
+ optimizingService().executeRemoval(optimizer.getToken());
+
+ Assertions.assertTrue(
+ optimizingService().isDraining(optimizer.getToken()),
+ "drain state must survive the failed release for a later retry");
+ Assertions.assertEquals(
+ 1,
+ optimizerManager().listOptimizers(group.getName()).size(),
+ "the optimizer must stay registered while its pod may still exist");
+
+ mockContainer.setReleaseAvailable(true);
+ optimizingService().executeRemoval(optimizer.getToken());
+
Assertions.assertFalse(optimizingService().isDraining(optimizer.getToken()));
+
Assertions.assertTrue(optimizerManager().listOptimizers(group.getName()).isEmpty());
+ }
+
+ /**
+ * DRA group whose rounds are driven manually with injected times. The huge
real cadence keeps the
+ * live keeper's own rounds from racing the injected ones on the per-group
decision state
+ * (idle-timeout 1200s respects the sustained ≤ idle/2 validation).
+ */
+ private ResourceGroup buildSlowDraResourceGroup(String groupName, int
minParallelism) {
+ this.currentGroupName = groupName;
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(OptimizerProperties.DYNAMIC_ALLOCATION_ENABLED, "true");
+ properties.put(
+ OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM,
String.valueOf(minParallelism));
+ properties.put(OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM,
"8");
+
properties.put(OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_PARALLELISM,
"1");
+
properties.put(OptimizerProperties.DYNAMIC_ALLOCATION_SUSTAINED_BACKLOG_TIMEOUT,
"600s");
+
properties.put(OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_IDLE_TIMEOUT,
"1200s");
+ properties.put("memory", "1024");
+ return new ResourceGroup.Builder(groupName, MOCK_CONTAINER_NAME)
+ .addProperties(properties)
+ .build();
+ }
+
+ /** Register an optimizer the way a booted pod would: persisted resource row
+ self-register. */
+ private OptimizerInstance registerOptimizer(String groupName, int
threadCount) {
+ org.apache.amoro.resource.Resource resource =
+ new org.apache.amoro.resource.Resource.Builder(
+ MOCK_CONTAINER_NAME, groupName,
org.apache.amoro.resource.ResourceType.OPTIMIZER)
+ .setThreadCount(threadCount)
+ .build();
+ optimizerManager().createResource(resource);
+ org.apache.amoro.api.OptimizerRegisterInfo registerInfo =
+ new org.apache.amoro.api.OptimizerRegisterInfo();
+ Map<String, String> registerProperties = Maps.newHashMap();
+ registerProperties.put(OptimizerProperties.OPTIMIZER_HEART_BEAT_INTERVAL,
"100");
+ registerInfo.setProperties(registerProperties);
+ registerInfo.setThreadCount(threadCount);
+ registerInfo.setMemoryMb(1024);
+ registerInfo.setGroupName(groupName);
+ registerInfo.setResourceId(resource.getResourceId());
+ registerInfo.setStartTime(System.currentTimeMillis());
+ String token = optimizingService().authenticate(registerInfo);
+ return optimizerManager().listOptimizers(groupName).stream()
+ .filter(optimizer -> token.equals(optimizer.getToken()))
+ .findFirst()
+ .orElseThrow(() -> new IllegalStateException("registered optimizer not
listed"));
+ }
+
+ /** An instance idle past executor-idle-timeout is drained and, being idle,
removed in-round. */
+ @Test
+ public void testIdleOptimizerScaledDownViaInjectedRounds() {
+ resourceAvailable.set(true);
+ scaleOutCallCount.set(0);
+ ResourceGroup group = buildSlowDraResourceGroup(TEST_GROUP_NAME + "-8", 0);
+ optimizerManager().createResourceGroup(group);
+ optimizingService().createResourceGroup(group);
+ OptimizerInstance optimizer = registerOptimizer(group.getName(), 1);
+
+ long t0 = System.currentTimeMillis();
+ optimizingService().evaluateDynamicAllocation(group.getName(), t0); //
seeds the observation
+ Assertions.assertEquals(1,
optimizerManager().listOptimizers(group.getName()).size());
+
+ optimizingService().evaluateDynamicAllocation(group.getName(), t0 +
1_300_000L);
+ Assertions.assertTrue(
+ optimizerManager().listOptimizers(group.getName()).isEmpty(),
+ "an idle instance past the timeout should be drained and removed in
the same round");
+ Assertions.assertTrue(
+ mockContainer.getReleasedResources().stream()
+ .anyMatch(r ->
optimizer.getResourceId().equals(r.getResourceId())));
+ }
+
+ /** The min-parallelism floor keeps the last instance no matter how long it
idles. */
+ @Test
+ public void testScaleDownRespectsFloor() {
+ resourceAvailable.set(true);
+ scaleOutCallCount.set(0);
+ ResourceGroup group = buildSlowDraResourceGroup(TEST_GROUP_NAME + "-9", 1);
+ optimizerManager().createResourceGroup(group);
+ optimizingService().createResourceGroup(group);
+ registerOptimizer(group.getName(), 1);
+
+ long t0 = System.currentTimeMillis();
+ optimizingService().evaluateDynamicAllocation(group.getName(), t0);
+ optimizingService().evaluateDynamicAllocation(group.getName(), t0 +
1_300_000L);
+
+ Assertions.assertEquals(
+ 1,
+ optimizerManager().listOptimizers(group.getName()).size(),
+ "the floor must keep the last instance");
+ }
+
+ /** Removals proceed one instance per cooldown period, never in batches. */
+ @Test
+ public void testScaleDownRemovesOneInstancePerCooldown() {
+ resourceAvailable.set(true);
+ scaleOutCallCount.set(0);
+ ResourceGroup group = buildSlowDraResourceGroup(TEST_GROUP_NAME + "-10",
0);
+ optimizerManager().createResourceGroup(group);
+ optimizingService().createResourceGroup(group);
+ registerOptimizer(group.getName(), 1);
+ registerOptimizer(group.getName(), 1);
+
+ long t0 = System.currentTimeMillis();
+ optimizingService().evaluateDynamicAllocation(group.getName(), t0);
+ long firstRemovalAt = t0 + 1_300_000L;
+ optimizingService().evaluateDynamicAllocation(group.getName(),
firstRemovalAt);
+ Assertions.assertEquals(
+ 1,
+ optimizerManager().listOptimizers(group.getName()).size(),
+ "only one instance per round may be removed");
+
+ // Inside the scale-down-cooldown window (default 1min): the second
instance stays.
+ optimizingService().evaluateDynamicAllocation(group.getName(),
firstRemovalAt + 30_000L);
+ Assertions.assertEquals(1,
optimizerManager().listOptimizers(group.getName()).size());
+
+ optimizingService().evaluateDynamicAllocation(group.getName(),
firstRemovalAt + 70_000L);
+ Assertions.assertTrue(
+ optimizerManager().listOptimizers(group.getName()).isEmpty(),
+ "the cooldown expiry should admit the next removal");
+ }
+
+ /**
+ * Disabling dynamic allocation mid-drain re-admits the draining pod to task
assignment: once the
+ * legacy floor keeper resumes duty for the group, a leftover drain block
would starve the pod
+ * forever.
+ */
+ @Test
+ public void testDisablingDraCancelsLingeringDrain() throws
InterruptedException {
+ resourceAvailable.set(true);
+ scaleOutCallCount.set(0);
+ ResourceGroup group = buildDraResourceGroup(TEST_GROUP_NAME + "-11", 0, 1);
+ optimizerManager().createResourceGroup(group);
+ optimizingService().createResourceGroup(group);
+ OptimizerInstance optimizer = registerOptimizer(group.getName(), 1);
+
+ // A drain that cannot complete (release keeps failing) lingers across
keeper rounds.
+ mockContainer.setReleaseAvailable(false);
+ optimizingService().beginGracefulDrain(optimizer.getToken(),
Long.MAX_VALUE);
+ Thread.sleep(200);
+
Assertions.assertTrue(optimizingService().isDraining(optimizer.getToken()));
+
+ Map<String, String> legacyProps = Maps.newHashMap();
+ legacyProps.put("memory", "1024");
+ ResourceGroup disabled =
+ new ResourceGroup.Builder(group.getName(), MOCK_CONTAINER_NAME)
+ .addProperties(legacyProps)
+ .build();
+ optimizerManager().updateResourceGroup(disabled);
+ optimizingService().updateResourceGroup(disabled);
+ Thread.sleep(500);
+
+ Assertions.assertFalse(
+ optimizingService().isDraining(optimizer.getToken()),
+ "unwatching a disabled group must lift its drain blocks");
+ Assertions.assertEquals(
+ 1,
+ optimizerManager().listOptimizers(group.getName()).size(),
+ "the pod survives and resumes normal duty");
+ }
}
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java
index 78a9e22e3..938b97783 100644
---
a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java
@@ -225,6 +225,71 @@ public class TestOptimizingQueue extends AMSTableTestBase {
queue.dispose();
}
+ @Test
+ public void testCollectDynamicAllocationLoadInFlightByToken() {
+ DefaultTableRuntime tableRuntime = initTableWithPartitionedFiles();
+ OptimizingQueue queue =
+ new OptimizingQueue(
+ CATALOG_MANAGER,
+ testResourceGroup(),
+ resourceGroup -> 2,
+ planExecutor,
+ Collections.singletonList(tableRuntime),
+ 1);
+ OptimizerThread threadA =
+ new OptimizerThread(1, null) {
+ @Override
+ public String getToken() {
+ return "token-a";
+ }
+ };
+ OptimizerThread threadB =
+ new OptimizerThread(2, null) {
+ @Override
+ public String getToken() {
+ return "token-b";
+ }
+ };
+
+ // One task SCHEDULED on token-a while the rest stay PLANNED: only the
occupying token is
+ // counted — PLANNED tasks carry no assignment and must not appear in the
map.
+ TaskRuntime<?> task = queue.pollTask(threadA, MAX_POLLING_TIME);
+ Assert.assertNotNull(task);
+ Assert.assertEquals(
+ ImmutableMap.of("token-a", 1),
queue.collectDynamicAllocationLoad().getInFlightByToken());
+
+ // ACKED still occupies the thread, so the token stays counted.
+ queue.ackTask(task.getTaskId(), threadA);
+ Assert.assertEquals(
+ ImmutableMap.of("token-a", 1),
queue.collectDynamicAllocationLoad().getInFlightByToken());
+
+ // A second optimizer polling the remaining task is aggregated under its
own token.
+ TaskRuntime<?> task2 = queue.pollTask(threadB, MAX_POLLING_TIME, true);
+ Assert.assertNotNull(task2);
+ Assert.assertEquals(
+ ImmutableMap.of("token-a", 1, "token-b", 1),
+ queue.collectDynamicAllocationLoad().getInFlightByToken());
+ queue.dispose();
+ }
+
+ @Test
+ public void testCollectDynamicAllocationLoadRecoversTaskTokens() {
+ DefaultTableRuntime tableRuntime = initTableWithFiles();
+ OptimizingQueue queue = buildOptimizingGroupService(tableRuntime);
+ TaskRuntime<?> task = queue.pollTask(optimizerThread, MAX_POLLING_TIME);
+ Assert.assertNotNull(task);
+ Assert.assertEquals(TaskRuntime.Status.SCHEDULED, task.getStatus());
+ queue.dispose();
+
+ // Rebuild the queue from persistent state, as an AMS restart does: the
recovered SCHEDULED
+ // task keeps its token, so the very first snapshot is accurate without
any rebuild code.
+ OptimizingQueue restored = buildOptimizingGroupService(tableRuntime);
+ DynamicAllocationState.GroupLoad load =
restored.collectDynamicAllocationLoad();
+ Assert.assertEquals(1, load.getBusyThreads());
+ Assert.assertEquals(ImmutableMap.of(optimizerThread.getToken(), 1),
load.getInFlightByToken());
+ restored.dispose();
+ }
+
@Test
public void testPollTaskWithOverQuotaDisabled() {
DefaultTableRuntime tableRuntime = initTableWithPartitionedFiles();
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestComputeScaleDown.java
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestComputeScaleDown.java
new file mode 100644
index 000000000..a8e473bea
--- /dev/null
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestComputeScaleDown.java
@@ -0,0 +1,208 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.amoro.server.optimizing.dra;
+
+import org.apache.amoro.OptimizerProperties;
+import org.apache.amoro.resource.ResourceGroup;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Scale-down decision tests: idle observation from snapshots, longest-idle
candidate selection, the
+ * min-parallelism floor against registered-minus-draining threads, and
cooldown rate limiting.
+ */
+public class TestComputeScaleDown {
+
+ private static final long T0 = 0L;
+ private static final long IDLE_MS = 300_000L; // executor-idle-timeout
default 5min
+ private static final long COOLDOWN_MS = 60_000L; // scale-down-cooldown
default 1min
+
+ private DynamicAllocationConfig config(int minParallelism) {
+ Map<String, String> props = new HashMap<>();
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_ENABLED, "true");
+ props.put(
+ OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM,
String.valueOf(minParallelism));
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM, "100");
+ return DynamicAllocationConfig.parse(
+ new ResourceGroup.Builder("group1",
"flink").addProperties(props).build());
+ }
+
+ private DynamicAllocationState.RemovalCandidate candidate(String token, int
threadCount) {
+ return new DynamicAllocationState.RemovalCandidate(token, threadCount);
+ }
+
+ private void observeIdle(DynamicAllocationState state, long now, String...
tokens) {
+ state.observe(
+ Arrays.stream(tokens).collect(java.util.stream.Collectors.toSet()),
+ Collections.emptyMap(),
+ now);
+ }
+
+ // --- idle qualification ---
+
+ @Test
+ void newTokenIsIdleFromFirstObservation() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ observeIdle(state, T0, "a");
+ List<DynamicAllocationState.RemovalCandidate> candidates =
+ Collections.singletonList(candidate("a", 1));
+
+ Assertions.assertNull(state.computeScaleDown(candidates, 1, 0, config(0),
T0 + IDLE_MS - 1));
+ Assertions.assertEquals("a", state.computeScaleDown(candidates, 1, 0,
config(0), T0 + IDLE_MS));
+ }
+
+ @Test
+ void busyObservationResetsIdleClock() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ observeIdle(state, T0, "a");
+ long t1 = T0 + 100_000;
+ state.observe(Collections.singleton("a"), Collections.singletonMap("a",
1), t1); // busy at t1
+ List<DynamicAllocationState.RemovalCandidate> candidates =
+ Collections.singletonList(candidate("a", 1));
+
+ Assertions.assertNull(state.computeScaleDown(candidates, 1, 0, config(0),
T0 + IDLE_MS));
+ Assertions.assertEquals("a", state.computeScaleDown(candidates, 1, 0,
config(0), t1 + IDLE_MS));
+ }
+
+ @Test
+ void neverObservedTokenIsNotSelected() {
+ // A candidate the keeper has not observed yet must not be treated as
long-idle.
+ DynamicAllocationState state = new DynamicAllocationState();
+ Assertions.assertNull(
+ state.computeScaleDown(
+ Collections.singletonList(candidate("ghost", 1)), 1, 0, config(0),
T0 + IDLE_MS));
+ }
+
+ @Test
+ void reRegisteredTokenIsSeededFresh() {
+ // Unregistration prunes the observation; a token that comes back (same
optimizer identity
+ // reused) must re-earn its idle time instead of inheriting the stale
pre-prune timestamp.
+ DynamicAllocationState state = new DynamicAllocationState();
+ observeIdle(state, T0, "a");
+ observeIdle(state, T0 + 10_000); // "a" unregistered: pruned
+ long t2 = T0 + 20_000;
+ observeIdle(state, t2, "a"); // back again
+ List<DynamicAllocationState.RemovalCandidate> candidates =
+ Collections.singletonList(candidate("a", 1));
+
+ Assertions.assertNull(state.computeScaleDown(candidates, 1, 0, config(0),
T0 + IDLE_MS));
+ Assertions.assertEquals("a", state.computeScaleDown(candidates, 1, 0,
config(0), t2 + IDLE_MS));
+ }
+
+ // --- candidate selection ---
+
+ @Test
+ void longestIdleCandidateWins() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ observeIdle(state, T0, "a", "b");
+ long t1 = T0 + 50_000;
+ state.observe(
+ Arrays.stream(new String[] {"a",
"b"}).collect(java.util.stream.Collectors.toSet()),
+ Collections.singletonMap("a", 1), // a busy at t1, b idle since T0
+ t1);
+
+ Assertions.assertEquals(
+ "b",
+ state.computeScaleDown(
+ Arrays.asList(candidate("a", 1), candidate("b", 1)), 2, 0,
config(0), t1 + IDLE_MS));
+ }
+
+ @Test
+ void onlyOneCandidatePerRound() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ observeIdle(state, T0, "a", "b");
+ String first =
+ state.computeScaleDown(
+ Arrays.asList(candidate("a", 1), candidate("b", 1)), 2, 0,
config(0), T0 + IDLE_MS);
+ Assertions.assertNotNull(first);
+ // The very next call within the cooldown window returns nothing, even
though the other
+ // instance is equally idle: one removal per cooldown period.
+ Assertions.assertNull(
+ state.computeScaleDown(
+ Arrays.asList(candidate("a", 1), candidate("b", 1)),
+ 2,
+ 1,
+ config(0),
+ T0 + IDLE_MS + 1));
+ }
+
+ // --- floor enforcement ---
+
+ @Test
+ void floorBlocksRemoval() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ observeIdle(state, T0, "a");
+ Assertions.assertNull(
+ state.computeScaleDown(
+ Collections.singletonList(candidate("a", 1)), 2, 0, config(2), T0
+ IDLE_MS));
+ }
+
+ @Test
+ void floorCountsDrainingThreadsAsAlreadyGone() {
+ // registered=3 still includes a draining instance; treating it as
capacity would let a second
+ // removal pass the floor check and land the group below min-parallelism
once both complete.
+ DynamicAllocationState state = new DynamicAllocationState();
+ observeIdle(state, T0, "b");
+ Assertions.assertNull(
+ state.computeScaleDown(
+ Collections.singletonList(candidate("b", 1)), 3, 1, config(2), T0
+ IDLE_MS));
+ }
+
+ @Test
+ void nextIdleCandidateIsPickedWhenLongestViolatesFloor() {
+ // Heterogeneous thread counts: removing the longest-idle 3-thread
instance would break the
+ // floor, but the shorter-idle 1-thread instance fits — pick it instead of
returning null.
+ DynamicAllocationState state = new DynamicAllocationState();
+ observeIdle(state, T0, "big");
+ long t1 = T0 + 50_000;
+ observeIdle(state, t1, "big", "small");
+
+ Assertions.assertEquals(
+ "small",
+ state.computeScaleDown(
+ Arrays.asList(candidate("big", 3), candidate("small", 1)),
+ 4,
+ 0,
+ config(2),
+ t1 + IDLE_MS));
+ }
+
+ // --- cooldown ---
+
+ @Test
+ void cooldownRateLimitsRemovals() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ observeIdle(state, T0, "a", "b");
+ List<DynamicAllocationState.RemovalCandidate> candidates =
+ Arrays.asList(candidate("a", 1), candidate("b", 1));
+
+ long firstAt = T0 + IDLE_MS;
+ Assertions.assertNotNull(state.computeScaleDown(candidates, 2, 0,
config(0), firstAt));
+ Assertions.assertNull(
+ state.computeScaleDown(candidates, 2, 1, config(0), firstAt +
COOLDOWN_MS - 1));
+ Assertions.assertNotNull(
+ state.computeScaleDown(candidates, 2, 1, config(0), firstAt +
COOLDOWN_MS));
+ }
+}
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestComputeScaleUp.java
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestComputeScaleUp.java
index d5b08ec56..58e852f0f 100644
---
a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestComputeScaleUp.java
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestComputeScaleUp.java
@@ -228,4 +228,30 @@ public class TestComputeScaleUp {
Assertions.assertEquals(
1, state.computeScaleUp(8, 8, 0, 10, config, T0 + BACKLOG_MS + 2 *
SUSTAINED_MS));
}
+
+ // --- demand-signal exposure for scale-down mutual exclusion ---
+
+ @Test
+ void demandActiveWhileBacklogGateStillHolds() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ // Demand exists but the backlog gate has not elapsed: no instances are
added yet, but the
+ // signal must read active — scaling down a warm instance right before the
gate opens would
+ // remove exactly the capacity the next round re-requests.
+ Assertions.assertEquals(0, state.computeScaleUp(2, 2, 3, 0, config(0, 100,
1), T0));
+ Assertions.assertTrue(state.wasDemandActive());
+ }
+
+ @Test
+ void demandInactiveOnQuietRound() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ Assertions.assertEquals(0, state.computeScaleUp(4, 2, 0, 0, config(0, 100,
2), T0));
+ Assertions.assertFalse(state.wasDemandActive());
+ }
+
+ @Test
+ void floorDeficitCountsAsDemandActive() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ Assertions.assertEquals(3, state.computeScaleUp(0, 0, 0, 0, config(5, 100,
2), T0));
+ Assertions.assertTrue(state.wasDemandActive());
+ }
}
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestDynamicAllocationConfig.java
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestDynamicAllocationConfig.java
index 66d5aa41f..e0eec2a03 100644
---
a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestDynamicAllocationConfig.java
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestDynamicAllocationConfig.java
@@ -195,6 +195,25 @@ public class TestDynamicAllocationConfig {
Assertions.assertThrows(IllegalArgumentException.class, () ->
parseAndValidate(group(props)));
}
+ @Test
+ void sustainedBacklogTimeoutAboveHalfIdleTimeoutIsRejected() {
+ // The scale keeper's evaluation cadence is sustained-backlog-timeout,
which is also the idle
+ // observation resolution: sampling slower than half the idle timeout lets
an instance that
+ // worked between samples be misjudged as continuously idle and drained.
+ Map<String, String> props = enabledProps();
+
props.put(OptimizerProperties.DYNAMIC_ALLOCATION_SUSTAINED_BACKLOG_TIMEOUT,
"200s");
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_IDLE_TIMEOUT,
"300s");
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
parseAndValidate(group(props)));
+ }
+
+ @Test
+ void sustainedBacklogTimeoutAtHalfIdleTimeoutIsAccepted() {
+ Map<String, String> props = enabledProps();
+
props.put(OptimizerProperties.DYNAMIC_ALLOCATION_SUSTAINED_BACKLOG_TIMEOUT,
"150s");
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_IDLE_TIMEOUT,
"300s");
+ assertDoesNotThrow(() -> parseAndValidate(group(props)));
+ }
+
@Test
void unparsableDurationIsRejected() {
Map<String, String> props = enabledProps();
diff --git a/docs/admin-guides/managing-optimizers.md
b/docs/admin-guides/managing-optimizers.md
index 5abc4faf1..336aec73a 100644
--- a/docs/admin-guides/managing-optimizers.md
+++ b/docs/admin-guides/managing-optimizers.md
@@ -296,9 +296,9 @@ The optimizer group supports the following properties:
| dynamic-allocation.executor-parallelism | All | No | 1
| Threads
per optimizer instance created by dynamic allocation (the scaling unit, like
Spark's `spark.executor.cores`). The floor and the cap must be reachable in
units of this size. For Kubernetes groups a value of 4–8 is recommended so
per-pod JVM overhead is shared across threads.
[...]
| dynamic-allocation.scheduler-backlog-timeout | All | No | 1min
| How
long optimizing demand must persist before the first scale-out.
[...]
| dynamic-allocation.sustained-backlog-timeout | All | No | 30s
|
Interval between subsequent scale-outs while demand persists; also the group's
scale evaluation cadence.
[...]
-| dynamic-allocation.executor-idle-timeout | All | No | 5min
| Idle
duration before an optimizer becomes a scale-down candidate (minimum 30s).
Scale-down lands in a later release.
[...]
-| dynamic-allocation.scale-down-cooldown | All | No | 1min
| Minimum
interval between scale-down removals. Scale-down lands in a later release.
[...]
-| dynamic-allocation.drain-timeout | All | No | 15min
|
Force-removal safety net for a draining optimizer during scale-down. Scale-down
lands in a later release.
[...]
+| dynamic-allocation.executor-idle-timeout | All | No | 5min
| Idle
duration before an optimizer becomes a scale-down candidate (minimum 30s). Must
be at least twice `dynamic-allocation.sustained-backlog-timeout`, since idle
time is observed once per scale evaluation round. At most one optimizer is
scaled down per round, and only while the group has no scale-up demand.
[...]
+| dynamic-allocation.scale-down-cooldown | All | No | 1min
| Minimum
interval between scale-down removals.
[...]
+| dynamic-allocation.drain-timeout | All | No | 15min
|
Force-removal safety net for a draining optimizer. A scaled-down optimizer is
drained gracefully: it stops receiving new tasks and is removed once its
in-flight tasks finish. If draining takes longer than this timeout, the
optimizer is removed anyway and its remaining tasks are re-executed on other
optimizers. [...]
| memory | Local | Yes | N/A
| The max
memory of JVM for local optimizer, in MBs.
[...]
| flink-conf.\<key\> | Flink | No | N/A
| Any
flink config options could be overwritten, priority is optimizing-group >
optimizing-container > flink-conf.yaml.
[...]
| spark-conf.\<key\> | Spark | No | N/A
| Any
spark config options could be overwritten, priority is optimizing-group >
optimizing-container > spark-defaults.conf.
[...]
