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 d29cbf4bc [AMORO-4271] AIP-5 Phase 2: demand-driven scale-up for
dynamic allocation groups (#4272)
d29cbf4bc is described below
commit d29cbf4bcfba2ee60b24c0708aa5e7a1962205b0
Author: Jiwon Park <[email protected]>
AuthorDate: Wed Jul 22 16:06:17 2026 +0900
[AMORO-4271] AIP-5 Phase 2: demand-driven scale-up for dynamic allocation
groups (#4272)
* [AMORO-4271] AIP-5 Phase 2: add dynamic-allocation.executor-parallelism
config
The scaling unit of dynamic allocation is one homogeneous K-thread
optimizer instance (the Spark executor model). K is configured by the
new dynamic-allocation.executor-parallelism property (default 1).
validate() rejects K < 1 and K > max-parallelism: a single K-thread
instance already exceeding the cap could never be created, which would
leave an enabled group as a silent no-op.
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4271] AIP-5 Phase 2: pure demand accounting (serviceable planned,
thread occupancy)
serviceablePlannedCount implements quota-mode-aware demand counting:
a proportional quota (targetQuota <= 1) scales with availableCore, so
the whole backlog is serviceable; an absolute quota (> 1) is a fixed
limit that scaling cannot raise, so only free slots count.
occupiesThread counts SCHEDULED as occupying: a thread is busy from
assignment (pollTask), not from ack; counting only ACKED would
overestimate headroom during the poll-to-ack window.
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4271] AIP-5 Phase 2: computeScaleUp scale-out decision
Per-group decision state (backlog timer, cadence, exponential ramp)
with injected time, so every scenario is deterministic.
Ordered checks:
- min-parallelism floor: enforced immediately, no timing gate.
- Immediate demand (busy + serviceable > effective): exponential ramp
(1, 2, 4, 8) clamped to the actual need; the ramp resets when the
clamp binds (Spark addExecutors semantics) or when demand clears.
- Future demand (pending tables while all threads are busy, including
the zero-optimizer cold start where nothing polls and planning never
runs): a single probe instance. Pending tables are not quantified
demand before planning, so no exponential growth on this signal.
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 toward a scale-out.
The max-parallelism cap always wins.
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4271] AIP-5 Phase 2: pending registration accounting with boot
deadline
Registration is optimizer-driven (the pod self-registers after boot),
so scale-up counts requested-but-unregistered capacity to avoid
duplicate scale-outs during the boot window. Entries carry their own
boot deadline: a request that never registers (image pull failure,
exhausted ResourceQuota, crash loop) is pruned instead of freezing
scale-up below real demand forever. Heartbeat expiry cannot cover this
window because it only starts after registration.
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4271] AIP-5 Phase 2: dedicated scale keeper for dynamic-allocation
groups
DRA-enabled groups are taken over from the legacy floor keeper, whose
min-parallelism-check cadence (minutes, multiplied by consecutive
attempts) would render the DRA backlog timeouts (seconds) physically
unreachable. OptimizerScaleKeeper reuses the AbstractKeeper
infrastructure (HA leader gating, DelayQueue, lifecycle) and evaluates
each group at its own sustained-backlog-timeout; the legacy keeper
keeps watching DRA groups only to resume floor duty if DRA is disabled
later.
Scale-outs are executed in executor-parallelism-thread instance units
(computeScaleUp), with requested-but-unregistered capacity counted via
PendingRegistrations so a booting pod is not re-requested every round;
a synchronous request failure is dropped immediately and retried.
Groups are watched on create, startup load, and update (covering
enabling DRA on an existing group at runtime); deleted or disabled
groups drop out of the watch set on their next evaluation.
OptimizingQueue.collectDynamicAllocationLoad() snapshots the demand
side: busy threads (SCHEDULED and ACKED), quota-mode-aware serviceable
PLANNED tasks, and PENDING tables - the only signal observable on a
cold group with zero optimizers.
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4271] AIP-5 Phase 2: warn when a group is planning-bound instead
of scaling
Idle threads while tables wait as PENDING and no PLANNED tasks
materialize means the bottleneck is the serialized planning
(optimizer.max-planning-parallelism), not thread capacity: scaling out
would only add more idle threads. The scale keeper surfaces this as an
edge-triggered warning naming the config to raise, and does not scale.
A cold group (zero threads) stays the future-demand case.
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4271] AIP-5 Phase 2: distinguish the scale keeper's failure modes
The keeper's drop-out path treated every irregularity the same way,
which conflates situations that need opposite handling:
- A transient group-read failure (e.g. a database hiccup) is not a
deletion: there is no periodic re-watch, so dropping the group there
would silently disable its dynamic allocation until the next config
change. Keep the task alive and retry shortly.
- After unwatching a disabled group, re-read it once: an update
re-enabling DRA concurrently would have had its watch() call
swallowed by the still-present watched-set entry, orphaning the
group until its next change.
- An enabled group whose queue is momentarily absent (a delete/create
racing the config watcher) is transient too; unwatch-plus-rewatch
there would spin a delay-0 hot loop of DB reads.
- A scale-out that fails after requestResource succeeded has started a
real pod; erasing its boot-window entry would re-request a duplicate
next round. Only a failure before the request is dropped and retried.
- Disabling DRA keeps the boot-window accounting (re-enabling within
the window must not re-request the same capacity); deleting the
group drops everything immediately, so a same-name group created
before the next evaluation does not inherit phantom capacity.
- The planning-bound warning now counts registered threads only (a
booting pod's phantom capacity is not idle threads) and requires the
condition to persist across two consecutive evaluations, since a
single snapshot can hold transiently while planning is in flight.
Also pins in the integration test that registration clears the
boot-window accounting: double-counted capacity would suppress demand
scaling.
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4271] AIP-5 Phase 2: keep floor handling consistent across
validation and scaling
Two floor edge cases:
- A floor unreachable in executor-parallelism units (e.g. min=5, max=6,
K=4: covering the floor needs 2 instances = 8 threads > max) passed
validation and then sat permanently below its floor with no signal -
the same silent no-op the validation rules exist to prevent. Reject
it up front; the keeper-side cap clamp stays as defense in depth for
configs persisted before this rule.
- A floor deficit (optimizers died, or a new group) now resets the
demand-phase state: previously a demand phase before the deficit
left a passed cadence gate and a grown ramp behind, so the first
demand after recovery fired immediately and oversized instead of
re-proving backlog persistence.
Also pins that a fractional absolute quota truncates like the poll
gate does ((int) 2.5 = 2 slots), keeping the two accountings aligned.
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4271] AIP-5 Phase 2: snapshot table runtimes under the scheduling
lock
SchedulingPolicy's table runtime map is a plain HashMap whose
canonical accesses all hold tableLock; the dynamic-allocation load
snapshot iterated it lock-free from the scale keeper thread, so a
concurrent addTable/removeTable could throw
ConcurrentModificationException and skip that whole evaluation round.
Add a snapshot accessor that copies the runtimes under the lock and
use it for the load snapshot.
Also covers collectDynamicAllocationLoad with a queue-level test on a
real table: the PENDING table is the only demand signal before any
poll (the cold-start case), and a polled task occupies its thread from
SCHEDULED on.
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4271] AIP-5 Phase 2: document the dynamic-allocation group
properties
Add the dynamic-allocation.* properties to the optimizer group
property table, mark the flat min-parallelism row deprecated in favor
of the namespaced key, and recommend executor-parallelism 4-8 for
Kubernetes groups so per-pod JVM overhead is shared across threads.
The scale-down-related properties are documented as landing in a later
release.
Signed-off-by: Jiwon Park <[email protected]>
---------
Signed-off-by: Jiwon Park <[email protected]>
Co-authored-by: ZhouJinsong <[email protected]>
---
.../amoro/server/DefaultOptimizingService.java | 281 +++++++++++++++++++-
.../amoro/server/optimizing/OptimizingQueue.java | 42 +++
.../amoro/server/optimizing/SchedulingPolicy.java | 16 ++
.../optimizing/dra/DynamicAllocationConfig.java | 52 ++++
.../optimizing/dra/DynamicAllocationState.java | 206 +++++++++++++++
.../optimizing/dra/PendingRegistrations.java | 71 +++++
.../amoro/server/TestOptimizerScaleKeeper.java | 286 +++++++++++++++++++++
.../server/optimizing/TestOptimizingQueue.java | 26 ++
.../server/optimizing/dra/TestComputeScaleUp.java | 231 +++++++++++++++++
.../dra/TestDynamicAllocationConfig.java | 64 +++++
.../optimizing/dra/TestDynamicAllocationState.java | 175 +++++++++++++
.../optimizing/dra/TestPendingRegistrations.java | 87 +++++++
.../java/org/apache/amoro/OptimizerProperties.java | 6 +
docs/admin-guides/managing-optimizers.md | 11 +-
14 files changed, 1542 insertions(+), 12 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 52345d2cc..6831d57c7 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
@@ -46,6 +46,8 @@ 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.optimizing.dra.DynamicAllocationConfig;
+import org.apache.amoro.server.optimizing.dra.DynamicAllocationState;
+import org.apache.amoro.server.optimizing.dra.PendingRegistrations;
import org.apache.amoro.server.persistence.StatedPersistentBase;
import org.apache.amoro.server.persistence.mapper.OptimizerMapper;
import org.apache.amoro.server.persistence.mapper.ResourceMapper;
@@ -59,6 +61,7 @@ import org.apache.amoro.server.resource.QuotaProvider;
import org.apache.amoro.server.table.DefaultTableRuntime;
import org.apache.amoro.server.table.RuntimeHandlerChain;
import org.apache.amoro.server.table.TableService;
+import
org.apache.amoro.shade.guava32.com.google.common.annotations.VisibleForTesting;
import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
import org.apache.amoro.shade.guava32.com.google.common.collect.Sets;
import
org.apache.amoro.shade.guava32.com.google.common.util.concurrent.ThreadFactoryBuilder;
@@ -116,6 +119,8 @@ public class DefaultOptimizingService extends
StatedPersistentBase
private final OptimizerKeeper optimizerKeeper = new
OptimizerKeeper("optimizer-keeper-thread");
private final OptimizerGroupKeeper optimizerGroupKeeper =
new OptimizerGroupKeeper("optimizer-group-keeper-thread");
+ private final OptimizerScaleKeeper optimizerScaleKeeper =
+ new OptimizerScaleKeeper("optimizer-scale-keeper-thread");
private final OptimizingConfigWatcher optimizingConfigWatcher = new
OptimizingConfigWatcher();
private final CatalogManager catalogManager;
private final OptimizerManager optimizerManager;
@@ -193,6 +198,7 @@ public class DefaultOptimizingService extends
StatedPersistentBase
maxPlanningParallelism);
optimizingQueueByGroup.put(groupName, optimizingQueue);
optimizerGroupKeeper.keepInTouch(groupName, 1);
+ optimizerScaleKeeper.watch(group);
});
optimizers.forEach(optimizer -> registerOptimizer(optimizer, false));
// Avoid keeping the tables in processing/pending status forever in below
cases:
@@ -229,6 +235,7 @@ public class DefaultOptimizingService extends
StatedPersistentBase
authOptimizers.put(optimizer.getToken(), optimizer);
optimizingQueueByToken.put(optimizer.getToken(), optimizingQueue);
optimizerKeeper.keepInTouch(optimizer);
+ optimizerScaleKeeper.onOptimizerRegistered(optimizer);
}
private void unregisterOptimizer(String token) {
@@ -405,17 +412,25 @@ public class DefaultOptimizingService extends
StatedPersistentBase
String groupName = resourceGroup.getName();
optimizingQueueByGroup.put(groupName, optimizingQueue);
optimizerGroupKeeper.keepInTouch(groupName, 1);
+ optimizerScaleKeeper.watch(resourceGroup);
});
}
public void deleteResourceGroup(String groupName) {
OptimizingQueue optimizingQueue = optimizingQueueByGroup.remove(groupName);
optimizingQueue.dispose();
+ optimizerScaleKeeper.onGroupDeleted(groupName);
}
public void updateResourceGroup(ResourceGroup resourceGroup) {
Optional.ofNullable(optimizingQueueByGroup.get(resourceGroup.getName()))
.ifPresent(queue -> queue.updateOptimizerGroup(resourceGroup));
+ optimizerScaleKeeper.watch(resourceGroup);
+ }
+
+ @VisibleForTesting
+ int pendingScaleThreads(String groupName) {
+ return optimizerScaleKeeper.pendingThreads(groupName);
}
public void dispose() {
@@ -426,6 +441,7 @@ public class DefaultOptimizingService extends
StatedPersistentBase
optimizingQueueByGroup.values().forEach(OptimizingQueue::dispose);
optimizerKeeper.dispose();
optimizerGroupKeeper.dispose();
+ optimizerScaleKeeper.dispose();
tableHandlerChain.dispose();
optimizingQueueByGroup.clear();
optimizingQueueByToken.clear();
@@ -497,6 +513,7 @@ public class DefaultOptimizingService extends
StatedPersistentBase
.collect(Collectors.toList()));
optimizerKeeper.start();
optimizerGroupKeeper.start();
+ optimizerScaleKeeper.start();
optimizingConfigWatcher.start();
LOG.info("SuspendingDetector for Optimizer has been started.");
LOG.info("OptimizerManagementService initializing has completed");
@@ -900,6 +917,13 @@ public class DefaultOptimizingService extends
StatedPersistentBase
return;
}
+ if (DynamicAllocationConfig.isEffectivelyEnabled(resourceGroup)) {
+ // Dynamic allocation owns this group's floor and demand scaling (see
+ // OptimizerScaleKeeper); keep watching in case it is disabled later.
+ keepInTouch(resourceGroup.getName(), 1);
+ return;
+ }
+
int requiredCores = keepingTask.tryKeeping(resourceGroup);
if (requiredCores <= 0) {
LOG.debug(
@@ -910,17 +934,6 @@ public class DefaultOptimizingService extends
StatedPersistentBase
if (keepingTask.getAttempts() > groupMaxKeepingAttempts) {
int minParallelism = keepingTask.getMinParallelism(resourceGroup);
- if (DynamicAllocationConfig.isEffectivelyEnabled(resourceGroup)) {
- // Dynamic allocation owns scale decisions for the group; never
erode its
- // min-parallelism floor automatically.
- LOG.warn(
- "Resource Group:{}, creating optimizer {} times in a row,
optimizers still below min-parallel:{}; dynamic allocation is enabled so
min-parallel is kept",
- resourceGroup.getName(),
- keepingTask.getAttempts(),
- minParallelism);
- keepInTouch(resourceGroup.getName(), 1);
- return;
- }
LOG.warn(
"Resource Group:{}, creating optimizer {} times in a row,
optimizers still below min-parallel:{}, will reset min-parallel to {}",
resourceGroup.getName(),
@@ -957,4 +970,250 @@ public class DefaultOptimizingService extends
StatedPersistentBase
requiredCores);
}
}
+
+ private class DraScaleTask implements Delayed {
+
+ private final String groupName;
+ private final long readyTimeMs;
+
+ private DraScaleTask(String groupName, long delayMs) {
+ this.groupName = groupName;
+ this.readyTimeMs = System.currentTimeMillis() + delayMs;
+ }
+
+ @Override
+ public long getDelay(@NotNull TimeUnit unit) {
+ return unit.convert(readyTimeMs - System.currentTimeMillis(),
TimeUnit.MILLISECONDS);
+ }
+
+ @Override
+ public int compareTo(@NotNull Delayed other) {
+ return Long.compare(readyTimeMs, ((DraScaleTask) other).readyTimeMs);
+ }
+ }
+
+ /**
+ * Keeper owning both the floor and the demand scaling of
dynamic-allocation-enabled groups
+ * (AIP-5). It is separate from {@link OptimizerGroupKeeper}, whose
min-parallelism-check cadence
+ * (minutes, multiplied by attempts) would render the DRA backlog timeouts
(seconds) unreachable;
+ * a group's scale evaluations run at its own sustained-backlog-timeout
instead.
+ */
+ private class OptimizerScaleKeeper extends AbstractKeeper<DraScaleTask> {
+
+ // Must exceed a normal pod boot including image pull: evicting a
legitimately booting pod
+ // from the pending accounting would cause duplicate scale-outs, which is
worse than a few
+ // conservative rounds with phantom capacity.
+ private static final long BOOT_TIMEOUT_MS = 3 * 60 * 1000L;
+
+ // Retry delay after a transient resource-group read failure, when the
group's configured
+ // cadence is unknown because the group itself could not be loaded.
+ private static final long TRANSIENT_RETRY_DELAY_MS = 5_000L;
+
+ private final Map<String, DynamicAllocationState> scaleStates = new
ConcurrentHashMap<>();
+ private final Map<String, PendingRegistrations> pendingRegistrations =
+ new ConcurrentHashMap<>();
+ private final Set<String> watchedGroups = ConcurrentHashMap.newKeySet();
+ private final Map<String, Integer> planningBoundStreaks = new
ConcurrentHashMap<>();
+
+ public OptimizerScaleKeeper(String threadName) {
+ super(threadName);
+ }
+
+ /** Start watching a group if dynamic allocation is effectively enabled on
it. Idempotent. */
+ public void watch(ResourceGroup resourceGroup) {
+ if (!DynamicAllocationConfig.isEffectivelyEnabled(resourceGroup)) {
+ return;
+ }
+ if (watchedGroups.add(resourceGroup.getName())) {
+ suspendingQueue.add(new DraScaleTask(resourceGroup.getName(), 0));
+ }
+ }
+
+ /** Clear the boot-window accounting of a registered optimizer
(AMS-launched ones only). */
+ public void onOptimizerRegistered(OptimizerInstance optimizer) {
+ if (optimizer.getResourceId() == null) {
+ return;
+ }
+ PendingRegistrations pending =
pendingRegistrations.get(optimizer.getGroupName());
+ if (pending != null) {
+ pending.registered(optimizer.getResourceId());
+ }
+ }
+
+ private void unwatch(String groupName) {
+ watchedGroups.remove(groupName);
+ scaleStates.remove(groupName);
+ planningBoundStreaks.remove(groupName);
+ // 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.
+ }
+
+ /**
+ * Full cleanup on group deletion. Unlike a disable, a deleted group's
boot-window accounting
+ * must go too: leaving it would leak the entry and, if a group with the
same name is created
+ * before the next evaluation, suppress its scale-up with the old group's
phantom capacity.
+ */
+ public void onGroupDeleted(String groupName) {
+ unwatch(groupName);
+ pendingRegistrations.remove(groupName);
+ }
+
+ @Override
+ protected void processTask(DraScaleTask task) {
+ ResourceGroup resourceGroup;
+ try {
+ resourceGroup = optimizerManager.getResourceGroup(task.groupName);
+ } catch (Exception e) {
+ // A transient failure (e.g. a database hiccup) must not be treated as
deletion: there is
+ // no periodic re-watch, so dropping the group here would silently
disable its dynamic
+ // allocation until the next config change. Keep the task alive and
retry.
+ LOG.warn(
+ "Failed to load resource group {} for dynamic allocation, will
retry",
+ task.groupName,
+ e);
+ suspendingQueue.add(new DraScaleTask(task.groupName,
TRANSIENT_RETRY_DELAY_MS));
+ return;
+ }
+ if (resourceGroup == null ||
!DynamicAllocationConfig.isEffectivelyEnabled(resourceGroup)) {
+ // Deleted or disabled: stop watching; an update re-enabling DRA
re-watches the group.
+ unwatch(task.groupName);
+ // An update may have re-enabled the group between our read and the
unwatch, in which
+ // case its watch() call was swallowed by the still-present
watchedGroups entry:
+ // double-check on a fresh read so such a group is not orphaned until
its next change.
+ recheckAfterUnwatch(task.groupName);
+ return;
+ }
+ OptimizingQueue queue = optimizingQueueByGroup.get(task.groupName);
+ if (queue == null) {
+ // The group exists with DRA enabled but its queue is momentarily
absent (e.g. a
+ // delete/recreate racing the config watcher). Unwatch + rewatch here
would spin a
+ // delay-0 hot loop until the watcher recreates the queue; treat it as
transient.
+ suspendingQueue.add(new DraScaleTask(task.groupName,
TRANSIENT_RETRY_DELAY_MS));
+ return;
+ }
+ DynamicAllocationConfig config =
DynamicAllocationConfig.parse(resourceGroup);
+ try {
+ scaleIfNeeded(resourceGroup, queue, config);
+ } catch (Throwable t) {
+ LOG.error("Dynamic allocation scale evaluation failed for group {}",
task.groupName, t);
+ } finally {
+ suspendingQueue.add(
+ new DraScaleTask(task.groupName,
config.getSustainedBacklogTimeout().toMillis()));
+ }
+ }
+
+ /** Threads still expected to register for the group; testing hook for
boot accounting. */
+ private int pendingThreads(String groupName) {
+ PendingRegistrations pending = pendingRegistrations.get(groupName);
+ return pending == null ? 0 :
pending.pendingThreads(System.currentTimeMillis());
+ }
+
+ private void recheckAfterUnwatch(String groupName) {
+ try {
+ ResourceGroup fresh = optimizerManager.getResourceGroup(groupName);
+ if (fresh != null) {
+ watch(fresh);
+ }
+ } catch (Exception e) {
+ // The group became unreadable right after a successful read; its next
update watches it.
+ LOG.warn("Failed to re-check resource group {} after unwatch",
groupName, e);
+ }
+ }
+
+ /**
+ * Warn when the planning-bound state (idle threads, PENDING tables,
nothing PLANNED — the
+ * bottleneck is {@code optimizer.max-planning-parallelism}, so scaling
out would only add idle
+ * threads) persists across two consecutive evaluations. A single snapshot
can hold this
+ * condition transiently while planning is merely in flight, so one round
is not evidence;
+ * counting registered threads only keeps a booting pod's phantom capacity
from being mistaken
+ * for idle threads. Warns once per episode.
+ */
+ private void warnOnPlanningBoundTransition(
+ String groupName, int registeredThreads,
DynamicAllocationState.GroupLoad load) {
+ boolean planningBound =
+ DynamicAllocationState.isPlanningBound(
+ registeredThreads,
+ load.getBusyThreads(),
+ load.getServiceablePlanned(),
+ load.getPendingTables());
+ if (!planningBound) {
+ planningBoundStreaks.remove(groupName);
+ return;
+ }
+ int streak = planningBoundStreaks.merge(groupName, 1, Integer::sum);
+ if (streak == 2) {
+ LOG.warn(
+ "Resource group {} is planning-bound: {} idle thread(s) while {}
table(s) are "
+ + "PENDING and no tasks are PLANNED. Scaling out will not
help; consider "
+ + "raising {}.",
+ groupName,
+ registeredThreads - load.getBusyThreads(),
+ load.getPendingTables(),
+ AmoroManagementConf.OPTIMIZER_MAX_PLANNING_PARALLELISM.key());
+ }
+ }
+
+ private void scaleIfNeeded(
+ ResourceGroup resourceGroup, OptimizingQueue queue,
DynamicAllocationConfig config) {
+ 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();
+ warnOnPlanningBoundTransition(groupName, registeredThreads, load);
+ int addInstances =
+ state.computeScaleUp(
+ effectiveThreads,
+ load.getBusyThreads(),
+ load.getServiceablePlanned(),
+ load.getPendingTables(),
+ config,
+ now);
+ if (addInstances <= 0) {
+ return;
+ }
+ int threadsPerInstance = config.getExecutorParallelism();
+ LOG.info(
+ "Dynamic allocation scaling out group {}: {} instance(s) of {}
thread(s), effective threads {}",
+ groupName,
+ addInstances,
+ threadsPerInstance,
+ effectiveThreads);
+ for (int i = 0; i < addInstances; i++) {
+ Resource resource =
+ new Resource.Builder(resourceGroup.getContainer(), groupName,
ResourceType.OPTIMIZER)
+ .setProperties(resourceGroup.getProperties())
+ .setThreadCount(threadsPerInstance)
+ .build();
+ ResourceContainer resourceContainer =
Containers.get(resource.getContainerName());
+ pending.requested(resource.getResourceId(), threadsPerInstance, now);
+ boolean podRequested = false;
+ try {
+ ((AbstractOptimizerContainer)
resourceContainer).requestResource(resource);
+ podRequested = true;
+ optimizerManager.createResource(resource);
+ } catch (Throwable t) {
+ if (podRequested) {
+ // The pod was started; only its persistence failed. Keep the
pending accounting —
+ // the pod will self-register — instead of erasing it and
re-requesting a duplicate.
+ LOG.warn(
+ "Dynamic allocation scale-out of group {} requested resource
{} but failed to "
+ + "persist it",
+ groupName,
+ resource.getResourceId(),
+ t);
+ } else {
+ pending.failed(resource.getResourceId());
+ LOG.warn("Dynamic allocation scale-out failed for group {}",
groupName, t);
+ }
+ }
+ }
+ }
+ }
}
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 52eb46b4c..bbbd6b288 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
@@ -38,6 +38,7 @@ import org.apache.amoro.server.AmoroServiceConstants;
import org.apache.amoro.server.catalog.CatalogManager;
import org.apache.amoro.server.manager.MetricManager;
import org.apache.amoro.server.optimizing.TaskRuntime.Status;
+import org.apache.amoro.server.optimizing.dra.DynamicAllocationState;
import org.apache.amoro.server.persistence.OptimizingProcessState;
import org.apache.amoro.server.persistence.PersistentBase;
import org.apache.amoro.server.persistence.TaskFilesPersistence;
@@ -418,6 +419,47 @@ public class OptimizingQueue extends PersistentBase {
.collect(Collectors.toList());
}
+ /**
+ * Snapshot the demand-side load of this queue for dynamic allocation: busy
threads, serviceable
+ * PLANNED tasks (quota-mode aware, see {@link
DynamicAllocationState#serviceablePlannedCount}),
+ * and PENDING tables. PENDING tables are observable with zero optimizers,
which makes them the
+ * only scale-up signal on a cold group where nothing polls and planning
never runs.
+ */
+ public DynamicAllocationState.GroupLoad collectDynamicAllocationLoad() {
+ Map<Long, Integer> plannedByTable = Maps.newHashMap();
+ Map<Long, Integer> occupiedByTable = Maps.newHashMap();
+ int busyThreads = 0;
+ for (TaskRuntime<?> task : collectTasks()) {
+ if (DynamicAllocationState.occupiesThread(task.getStatus())) {
+ busyThreads++;
+ occupiedByTable.merge(task.getTableId(), 1, Integer::sum);
+ } else if (task.getStatus() == Status.PLANNED) {
+ plannedByTable.merge(task.getTableId(), 1, Integer::sum);
+ }
+ }
+ Map<Long, Double> targetQuotaByTable = Maps.newHashMap();
+ int pendingTables = 0;
+ for (DefaultTableRuntime tableRuntime : scheduler.snapshotTableRuntimes())
{
+ targetQuotaByTable.put(
+ tableRuntime.getTableIdentifier().getId(),
+ tableRuntime.getOptimizingConfig().getTargetQuota());
+ if (tableRuntime.getOptimizingStatus() == OptimizingStatus.PENDING) {
+ pendingTables++;
+ }
+ }
+ List<DynamicAllocationState.TableDemand> demands = Lists.newArrayList();
+ plannedByTable.forEach(
+ (tableId, planned) ->
+ demands.add(
+ new DynamicAllocationState.TableDemand(
+ planned,
+ // Unknown table (racing removal): default to
proportional, counting in full.
+ targetQuotaByTable.getOrDefault(tableId, 1.0),
+ occupiedByTable.getOrDefault(tableId, 0))));
+ return new DynamicAllocationState.GroupLoad(
+ busyThreads, DynamicAllocationState.serviceablePlannedCount(demands),
pendingTables);
+ }
+
public void retryTask(TaskRuntime<?> taskRuntime) {
findProcess(taskRuntime.getTaskId()).resetTask((TaskRuntime<RewriteStageTask>)
taskRuntime);
}
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/SchedulingPolicy.java
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/SchedulingPolicy.java
index 0759f1e36..8bb2d0eb1 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/SchedulingPolicy.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/SchedulingPolicy.java
@@ -28,9 +28,11 @@ import
org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Iterator;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.ServiceLoader;
@@ -146,4 +148,18 @@ public class SchedulingPolicy {
Map<ServerTableIdentifier, DefaultTableRuntime> getTableRuntimeMap() {
return tableRuntimeMap;
}
+
+ /**
+ * Copy the current table runtimes under {@code tableLock}. {@code
tableRuntimeMap} is a plain
+ * {@link HashMap} whose canonical accesses all hold the lock, so callers on
other threads (e.g.
+ * the dynamic-allocation scale keeper) must iterate a snapshot instead of
the live map.
+ */
+ public List<DefaultTableRuntime> snapshotTableRuntimes() {
+ tableLock.lock();
+ try {
+ return new ArrayList<>(tableRuntimeMap.values());
+ } finally {
+ tableLock.unlock();
+ }
+ }
}
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 ad62ad291..15c9de841 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
@@ -42,6 +42,7 @@ public class DynamicAllocationConfig {
private final boolean enabled;
private final Integer minParallelism;
private final Integer maxParallelism;
+ private final int executorParallelism;
private final Duration schedulerBacklogTimeout;
private final Duration sustainedBacklogTimeout;
private final Duration executorIdleTimeout;
@@ -54,6 +55,7 @@ public class DynamicAllocationConfig {
boolean enabled,
Integer minParallelism,
Integer maxParallelism,
+ int executorParallelism,
Duration schedulerBacklogTimeout,
Duration sustainedBacklogTimeout,
Duration executorIdleTimeout,
@@ -64,6 +66,7 @@ public class DynamicAllocationConfig {
this.enabled = enabled;
this.minParallelism = minParallelism;
this.maxParallelism = maxParallelism;
+ this.executorParallelism = executorParallelism;
this.schedulerBacklogTimeout = schedulerBacklogTimeout;
this.sustainedBacklogTimeout = sustainedBacklogTimeout;
this.executorIdleTimeout = executorIdleTimeout;
@@ -88,12 +91,19 @@ public class DynamicAllocationConfig {
PropertyUtil.propertyAsNullableInt(
properties,
OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM);
+ int executorParallelism =
+ PropertyUtil.propertyAsInt(
+ properties,
+ OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_PARALLELISM,
+
OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_PARALLELISM_DEFAULT);
+
return new DynamicAllocationConfig(
group.getName(),
group.getContainer(),
enabled,
minParallelism,
maxParallelism,
+ executorParallelism,
parseDuration(
properties,
OptimizerProperties.DYNAMIC_ALLOCATION_SCHEDULER_BACKLOG_TIMEOUT,
@@ -269,6 +279,44 @@ public class DynamicAllocationConfig {
maxParallelism,
OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM_LIMIT));
}
+ if (executorParallelism < 1) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Resource group:%s '%s'(%d) must be >= 1.",
+ groupName,
+ OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_PARALLELISM,
+ executorParallelism));
+ }
+ // A single executorParallelism-thread instance is the scaling unit; if it
alone exceeds
+ // max-parallelism, scale-up could never create anything, leaving a silent
no-op group.
+ if (executorParallelism > maxParallelism) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Resource group:%s '%s'(%d) must not exceed '%s'(%d).",
+ groupName,
+ OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_PARALLELISM,
+ executorParallelism,
+ OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM,
+ maxParallelism));
+ }
+ // The floor is satisfied in executor-parallelism-thread instance units;
if covering it would
+ // already exceed max-parallelism, the group would silently sit below its
floor forever.
+ int floorThreads =
+ (minParallelism + executorParallelism - 1) / executorParallelism *
executorParallelism;
+ if (floorThreads > maxParallelism) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Resource group:%s '%s'(%d) is not reachable in '%s'(%d) units:
covering the floor "
+ + "requires %d threads, exceeding '%s'(%d).",
+ groupName,
+ OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM,
+ minParallelism,
+ OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_PARALLELISM,
+ executorParallelism,
+ floorThreads,
+ OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM,
+ maxParallelism));
+ }
Duration idleMin =
ConfigHelpers.TimeUtils.parseDuration(
OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_IDLE_TIMEOUT_MIN);
@@ -340,6 +388,10 @@ public class DynamicAllocationConfig {
return maxParallelism;
}
+ public int getExecutorParallelism() {
+ return executorParallelism;
+ }
+
public Duration getSchedulerBacklogTimeout() {
return schedulerBacklogTimeout;
}
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
new file mode 100644
index 000000000..9ac84334a
--- /dev/null
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java
@@ -0,0 +1,206 @@
+/*
+ * 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.server.optimizing.TaskRuntime;
+
+import java.util.Collection;
+
+/**
+ * Per-group scale-up decision state for dynamic allocation (AIP-5): the
backlog timer, the
+ * scale-out cadence, and the exponential ramp. {@link #computeScaleUp} is
driven by injected inputs
+ * (loads, config, time), so the decision logic is deterministic and testable
without a live
+ * optimizing queue; the static demand-accounting helpers are pure functions.
+ */
+public final class DynamicAllocationState {
+
+ /** When demand was first observed; {@code -1} while there is no demand. */
+ private long backlogSinceMs = -1;
+
+ /** Earliest time the next scale-out may happen; {@code -1} before the first
one. */
+ private long nextAllowedAddMs = -1;
+
+ /** Instances to add in the next immediate-demand round (1, 2, 4, 8 ...). */
+ private int rampInstances = 1;
+
+ /**
+ * Decide how many executor-parallelism-thread optimizer instances to create
in this round.
+ *
+ * <p>Ordered checks: the {@code min-parallelism} floor is enforced
immediately (no timing gate);
+ * immediate demand ({@code busy + serviceable > effective}) scales
exponentially, clamped to the
+ * actual need (Spark semantics: the ramp resets when the clamp binds, and a
round with no demand
+ * resets it too); future demand (pending tables while every thread is busy
— including the
+ * zero-optimizer cold start, where nothing polls and planning never runs)
adds a single probe
+ * instance, because pending tables are not quantified demand before
planning. Demand must persist
+ * for {@code scheduler-backlog-timeout} before the first scale-out;
subsequent ones are spaced by
+ * {@code sustained-backlog-timeout}. The {@code max-parallelism} cap always
wins.
+ *
+ * @return the number of instances of {@code executor-parallelism} threads
to create, {@code >= 0}
+ */
+ public int computeScaleUp(
+ int effectiveThreads,
+ int busyThreads,
+ int serviceablePlanned,
+ int pendingTables,
+ DynamicAllocationConfig config,
+ long nowMs) {
+ int k = config.getExecutorParallelism();
+ int allowedInstances = Math.max(0, (config.getMaxParallelism() -
effectiveThreads) / k);
+
+ int minParallelism = config.getMinParallelism();
+ if (effectiveThreads < minParallelism) {
+ // 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.
+ backlogSinceMs = -1;
+ nextAllowedAddMs = -1;
+ rampInstances = 1;
+ int neededInstances = ceilDiv(minParallelism - effectiveThreads, k);
+ return Math.min(neededInstances, allowedInstances);
+ }
+
+ int actionableNeed = Math.max(busyThreads + serviceablePlanned -
effectiveThreads, 0);
+ boolean futureDemand = pendingTables > 0 && busyThreads >=
effectiveThreads;
+ if (actionableNeed <= 0 && !futureDemand) {
+ backlogSinceMs = -1;
+ nextAllowedAddMs = -1;
+ rampInstances = 1;
+ return 0;
+ }
+
+ if (backlogSinceMs < 0) {
+ backlogSinceMs = nowMs;
+ nextAllowedAddMs = -1;
+ }
+ long gate =
+ nextAllowedAddMs >= 0
+ ? nextAllowedAddMs
+ : backlogSinceMs + config.getSchedulerBacklogTimeout().toMillis();
+ if (nowMs < gate) {
+ return 0;
+ }
+
+ int add;
+ if (actionableNeed > 0) {
+ int wantInstances = ceilDiv(actionableNeed, k);
+ add = Math.min(Math.min(wantInstances, rampInstances), allowedInstances);
+ if (add <= 0) {
+ return 0;
+ }
+ // Spark semantics: keep doubling only while the ramp is the binding
constraint; once the
+ // actual need clamps the add, a grown ramp is no longer justified by
demand.
+ rampInstances = wantInstances > rampInstances ? rampInstances * 2 : 1;
+ } else {
+ add = Math.min(1, allowedInstances);
+ if (add <= 0) {
+ return 0;
+ }
+ rampInstances = 1;
+ }
+ nextAllowedAddMs = nowMs + config.getSustainedBacklogTimeout().toMillis();
+ return add;
+ }
+
+ private static int ceilDiv(int value, int divisor) {
+ return (value + divisor - 1) / divisor;
+ }
+
+ /** Snapshot of a group's current load, the demand-side inputs of {@link
#computeScaleUp}. */
+ public static class GroupLoad {
+ private final int busyThreads;
+ private final int serviceablePlanned;
+ private final int pendingTables;
+
+ public GroupLoad(int busyThreads, int serviceablePlanned, int
pendingTables) {
+ this.busyThreads = busyThreads;
+ this.serviceablePlanned = serviceablePlanned;
+ this.pendingTables = pendingTables;
+ }
+
+ public int getBusyThreads() {
+ return busyThreads;
+ }
+
+ public int getServiceablePlanned() {
+ return serviceablePlanned;
+ }
+
+ public int getPendingTables() {
+ return pendingTables;
+ }
+ }
+
+ /** Per-table demand snapshot consumed by {@link
#serviceablePlannedCount(Collection)}. */
+ public static class TableDemand {
+ private final int plannedCount;
+ private final double targetQuota;
+ private final int occupiedThreads;
+
+ public TableDemand(int plannedCount, double targetQuota, int
occupiedThreads) {
+ this.plannedCount = plannedCount;
+ this.targetQuota = targetQuota;
+ this.occupiedThreads = occupiedThreads;
+ }
+ }
+
+ /**
+ * Count the PLANNED tasks that adding optimizer capacity could actually
drain.
+ *
+ * <p>A table with a proportional quota ({@code targetQuota <= 1}) is
limited to {@code
+ * ceil(targetQuota * availableCore)} threads, so scaling up raises its
limit and the whole
+ * backlog is serviceable. A table with an absolute quota ({@code > 1}) has
a fixed thread limit
+ * that scaling cannot raise, so only its currently free slots are
serviceable.
+ */
+ public static int serviceablePlannedCount(Collection<TableDemand> demands) {
+ int total = 0;
+ for (TableDemand demand : demands) {
+ if (demand.targetQuota <= 1) {
+ total += demand.plannedCount;
+ } else {
+ int freeSlots = Math.max(0, (int) demand.targetQuota -
demand.occupiedThreads);
+ total += Math.min(demand.plannedCount, freeSlots);
+ }
+ }
+ return total;
+ }
+
+ /**
+ * Whether a task in this status occupies an optimizer thread. A thread is
occupied from
+ * assignment ({@code SCHEDULED}, set by {@code pollTask}) until the task
terminates; counting
+ * only {@code ACKED} would overestimate headroom during the poll-to-ack
window.
+ */
+ public static boolean occupiesThread(TaskRuntime.Status status) {
+ return status == TaskRuntime.Status.SCHEDULED || status ==
TaskRuntime.Status.ACKED;
+ }
+
+ /**
+ * Whether the group is bottlenecked on planning rather than on thread
capacity: threads sit idle
+ * while tables wait as PENDING and no PLANNED tasks materialize (planning
is serialized by {@code
+ * optimizer.max-planning-parallelism}). Scaling out in this state would
only add more idle
+ * threads, so the condition is surfaced as a warning instead of a
scale-out. A cold group (zero
+ * threads) is the future-demand case, not a planning bottleneck.
+ */
+ public static boolean isPlanningBound(
+ int effectiveThreads, int busyThreads, int serviceablePlanned, int
pendingTables) {
+ return effectiveThreads > 0
+ && busyThreads < effectiveThreads
+ && serviceablePlanned == 0
+ && pendingTables > 0;
+ }
+}
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/PendingRegistrations.java
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/PendingRegistrations.java
new file mode 100644
index 000000000..abc80c537
--- /dev/null
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/PendingRegistrations.java
@@ -0,0 +1,71 @@
+/*
+ * 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 java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Requested-but-not-yet-registered optimizer capacity of one resource group.
Registration is
+ * optimizer-driven — the pod self-registers after boot — so scale-up must
count capacity it has
+ * already requested to avoid duplicate scale-outs during the boot window, but
a request that never
+ * registers (image pull failure, exhausted ResourceQuota, crash loop) must
not count forever:
+ * heartbeat expiry only starts after registration, so entries carry their own
boot deadline.
+ * Thread-safe: the scale keeper and the registration RPC touch this
concurrently.
+ */
+public class PendingRegistrations {
+
+ private final long timeoutMs;
+ private final Map<String, Entry> entries = new ConcurrentHashMap<>();
+
+ private static class Entry {
+ private final int threads;
+ private final long deadlineMs;
+
+ private Entry(int threads, long deadlineMs) {
+ this.threads = threads;
+ this.deadlineMs = deadlineMs;
+ }
+ }
+
+ public PendingRegistrations(long timeoutMs) {
+ this.timeoutMs = timeoutMs;
+ }
+
+ /** Record a resource request whose optimizer has not registered yet. */
+ public void requested(String resourceId, int threads, long nowMs) {
+ entries.put(resourceId, new Entry(threads, nowMs + timeoutMs));
+ }
+
+ /** The optimizer of this resource registered; its capacity is now real. */
+ public void registered(String resourceId) {
+ entries.remove(resourceId);
+ }
+
+ /** The resource request failed synchronously; drop it immediately. */
+ public void failed(String resourceId) {
+ entries.remove(resourceId);
+ }
+
+ /** Total threads still expected to register, pruning entries past their
boot deadline. */
+ public int pendingThreads(long nowMs) {
+ entries.values().removeIf(entry -> nowMs >= entry.deadlineMs);
+ return entries.values().stream().mapToInt(entry -> entry.threads).sum();
+ }
+}
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
new file mode 100644
index 000000000..8a57acb6c
--- /dev/null
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerScaleKeeper.java
@@ -0,0 +1,286 @@
+/*
+ * 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;
+
+import org.apache.amoro.BasicTableTestHelper;
+import org.apache.amoro.OptimizerProperties;
+import org.apache.amoro.TableFormat;
+import org.apache.amoro.TableTestHelper;
+import org.apache.amoro.catalog.BasicCatalogTestHelper;
+import org.apache.amoro.catalog.CatalogTestHelper;
+import org.apache.amoro.resource.ResourceContainer;
+import org.apache.amoro.resource.ResourceGroup;
+import org.apache.amoro.server.resource.ContainerMetadata;
+import org.apache.amoro.server.resource.Containers;
+import org.apache.amoro.server.resource.OptimizerInstance;
+import org.apache.amoro.server.table.AMSTableTestBase;
+import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
+import org.apache.iceberg.common.DynFields;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.Test;
+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.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+/**
+ * Integration tests for the dynamic-allocation scale keeper: DRA-enabled
groups are taken over from
+ * the legacy floor keeper and scaled in executor-parallelism-thread instance
units, with
+ * requested-but-unregistered capacity preventing duplicate scale-outs during
the pod boot window.
+ */
+@RunWith(Parameterized.class)
+public class TestOptimizerScaleKeeper extends AMSTableTestBase {
+
+ private static final String TEST_GROUP_NAME = "test-scale-keeper-group";
+ private static final String MOCK_CONTAINER_NAME = "mock-scale-container";
+
+ private final AtomicBoolean resourceAvailable = new AtomicBoolean(true);
+ private final AtomicInteger scaleOutCallCount = new AtomicInteger(0);
+ // Null simulates a pod that was requested but never boots far enough to
self-register.
+ private volatile Function<org.apache.amoro.api.OptimizerRegisterInfo,
String> optimizerRegistrar;
+ private static boolean originIsInitialized = false;
+ private String currentGroupName;
+
+ public TestOptimizerScaleKeeper(
+ CatalogTestHelper catalogTestHelper, TableTestHelper tableTestHelper) {
+ super(catalogTestHelper, tableTestHelper, false);
+ }
+
+ @Parameterized.Parameters(name = "{0}, {1}")
+ public static Object[] parameters() {
+ return new Object[][] {
+ {new BasicCatalogTestHelper(TableFormat.ICEBERG), new
BasicTableTestHelper(true, false)}
+ };
+ }
+
+ @Before
+ public void prepare() throws Exception {
+ optimizerRegistrar = registerInfo ->
optimizingService().authenticate(registerInfo);
+ setupMockContainer(() -> currentGroupName);
+ }
+
+ @After
+ public void clear() {
+ if (currentGroupName == null) {
+ return;
+ }
+ try {
+ optimizerManager()
+ .listOptimizers(currentGroupName)
+ .forEach(
+ optimizer ->
+ optimizingService()
+ .deleteOptimizer(optimizer.getGroupName(),
optimizer.getResourceId()));
+ try {
+ optimizingService().deleteResourceGroup(currentGroupName);
+ } catch (Exception ignored) {
+ }
+ try {
+ optimizerManager().deleteResourceGroup(currentGroupName);
+ } catch (Exception ignored) {
+ }
+ } catch (Exception e) {
+ // ignore
+ } finally {
+ currentGroupName = null;
+ }
+ }
+
+ @AfterClass
+ public static void cleanup() {
+ if (!originIsInitialized) {
+ DynFields.UnboundField<Boolean> initializedField =
+ DynFields.builder().hiddenImpl(Containers.class,
"isInitialized").build();
+ initializedField.asStatic().set(false);
+ }
+ }
+
+ private void setupMockContainer(Supplier<String> targetGroupNameSupplier)
throws Exception {
+ TestOptimizerGroupKeeper.MockOptimizerContainer mockContainer =
+ new TestOptimizerGroupKeeper.MockOptimizerContainer(
+ resourceAvailable,
+ scaleOutCallCount,
+ registerInfo -> {
+ Function<org.apache.amoro.api.OptimizerRegisterInfo, String>
registrar =
+ optimizerRegistrar;
+ return registrar == null ? null : registrar.apply(registerInfo);
+ },
+ targetGroupNameSupplier);
+
+ DynFields.UnboundField<Boolean> initializedField =
+ DynFields.builder().hiddenImpl(Containers.class,
"isInitialized").build();
+ if (!initializedField.asStatic().get()) {
+ originIsInitialized = false;
+ initializedField.asStatic().set(true);
+ }
+
+ DynFields.UnboundField<Map<String, Object>> containersField =
+ DynFields.builder().hiddenImpl(Containers.class,
"globalContainers").build();
+ Map<String, Object> globalContainers = containersField.asStatic().get();
+
+ ContainerMetadata metadata =
+ new ContainerMetadata(
+ MOCK_CONTAINER_NAME,
TestOptimizerGroupKeeper.MockOptimizerContainer.class.getName());
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(OptimizerProperties.AMS_HOME, "/tmp");
+ properties.put(OptimizerProperties.AMS_OPTIMIZER_URI,
"thrift://localhost:1261");
+ properties.put("memory", "1024");
+ metadata.setProperties(properties);
+
+ Class<?> wrapperClass =
+
Class.forName("org.apache.amoro.server.resource.Containers$ContainerWrapper");
+ java.lang.reflect.Constructor<?> constructor =
+ wrapperClass.getDeclaredConstructor(ContainerMetadata.class,
ResourceContainer.class);
+ constructor.setAccessible(true);
+ Object wrapper = constructor.newInstance(metadata, mockContainer);
+ globalContainers.put(MOCK_CONTAINER_NAME, wrapper);
+ }
+
+ private ResourceGroup buildDraResourceGroup(String groupName, int
minParallelism, int k) {
+ 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,
String.valueOf(k));
+ // Fast timings so the scale keeper runs several rounds within a short
sleep.
+
properties.put(OptimizerProperties.DYNAMIC_ALLOCATION_SCHEDULER_BACKLOG_TIMEOUT,
"1ms");
+
properties.put(OptimizerProperties.DYNAMIC_ALLOCATION_SUSTAINED_BACKLOG_TIMEOUT,
"50ms");
+ properties.put("memory", "1024");
+ return new ResourceGroup.Builder(groupName, MOCK_CONTAINER_NAME)
+ .addProperties(properties)
+ .build();
+ }
+
+ /**
+ * The floor of a DRA group is satisfied in executor-parallelism-thread
instance units by the
+ * scale keeper, not by the legacy keeper's single deficit-sized instance:
min-parallelism=2 with
+ * executor-parallelism=1 must produce two 1-thread instances, not one
2-thread instance.
+ */
+ @Test
+ public void testDraFloorSatisfiedInExecutorParallelismUnits() throws
InterruptedException {
+ resourceAvailable.set(true);
+ scaleOutCallCount.set(0);
+ ResourceGroup group = buildDraResourceGroup(TEST_GROUP_NAME + "-1", 2, 1);
+
+ optimizerManager().createResourceGroup(group);
+ optimizingService().createResourceGroup(group);
+
+ Thread.sleep(500);
+
+ List<OptimizerInstance> optimizers =
optimizerManager().listOptimizers(group.getName());
+ Assertions.assertEquals(
+ 2, optimizers.size(), "floor should be satisfied by K-thread
instances");
+ optimizers.forEach(
+ optimizer ->
+ Assertions.assertEquals(
+ 1,
+ optimizer.getThreadCount(),
+ "each instance should have executor-parallelism threads"));
+ Assertions.assertEquals(
+ 0,
+ optimizingService().pendingScaleThreads(group.getName()),
+ "registration must clear the boot-window accounting, or registered
capacity would be "
+ + "double-counted and suppress demand scaling");
+ }
+
+ /**
+ * Requested-but-unregistered capacity counts toward the effective threads:
while pods are booting
+ * (never registering here), the scale keeper must not re-request the same
deficit every round the
+ * way the legacy keeper would.
+ */
+ @Test
+ public void testBootWindowPreventsDuplicateScaleOuts() throws
InterruptedException {
+ resourceAvailable.set(true);
+ scaleOutCallCount.set(0);
+ optimizerRegistrar = null; // pods are requested but never self-register
+ ResourceGroup group = buildDraResourceGroup(TEST_GROUP_NAME + "-2", 2, 1);
+
+ optimizerManager().createResourceGroup(group);
+ optimizingService().createResourceGroup(group);
+
+ // ~10 scale rounds at the 50ms cadence; without boot-window accounting
each round would
+ // re-request the full deficit.
+ Thread.sleep(500);
+
+ Assertions.assertEquals(
+ 2,
+ scaleOutCallCount.get(),
+ "the deficit must be requested exactly once while the pods are still
booting");
+ }
+
+ /**
+ * A synchronous scale-out failure must not stick as phantom pending
capacity: the failed request
+ * is dropped immediately and retried on a later round.
+ */
+ @Test
+ public void testFailedScaleOutIsRetriedNextRound() throws
InterruptedException {
+ resourceAvailable.set(false);
+ scaleOutCallCount.set(0);
+ ResourceGroup group = buildDraResourceGroup(TEST_GROUP_NAME + "-3", 1, 1);
+
+ optimizerManager().createResourceGroup(group);
+ optimizingService().createResourceGroup(group);
+
+ Thread.sleep(500);
+
+ Assertions.assertTrue(
+ scaleOutCallCount.get() >= 2,
+ "failed requests should be retried instead of freezing scale-up: "
+ + scaleOutCallCount.get());
+ }
+
+ /** Enabling DRA on an existing group at runtime brings it under the scale
keeper. */
+ @Test
+ public void testEnablingDraAtRuntimeBringsGroupUnderScaleKeeper() throws
InterruptedException {
+ resourceAvailable.set(true);
+ scaleOutCallCount.set(0);
+ this.currentGroupName = TEST_GROUP_NAME + "-4";
+ Map<String, String> legacyProps = Maps.newHashMap();
+ legacyProps.put("memory", "1024");
+ ResourceGroup legacyGroup =
+ new ResourceGroup.Builder(currentGroupName, MOCK_CONTAINER_NAME)
+ .addProperties(legacyProps)
+ .build();
+
+ optimizerManager().createResourceGroup(legacyGroup);
+ optimizingService().createResourceGroup(legacyGroup);
+ Thread.sleep(100);
+ Assertions.assertEquals(0, scaleOutCallCount.get(), "no demand and no
floor: no scale-out");
+
+ ResourceGroup draGroup = buildDraResourceGroup(currentGroupName, 2, 1);
+ optimizerManager().updateResourceGroup(draGroup);
+ optimizingService().updateResourceGroup(draGroup);
+
+ Thread.sleep(500);
+
+ List<OptimizerInstance> optimizers =
optimizerManager().listOptimizers(currentGroupName);
+ Assertions.assertEquals(
+ 2, optimizers.size(), "runtime-enabled DRA group should reach its
floor in K units");
+ }
+}
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 772f15c01..78a9e22e3 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
@@ -50,6 +50,7 @@ import org.apache.amoro.optimizing.TaskProperties;
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.dra.DynamicAllocationState;
import org.apache.amoro.server.resource.OptimizerInstance;
import org.apache.amoro.server.resource.OptimizerThread;
import org.apache.amoro.server.resource.QuotaProvider;
@@ -199,6 +200,31 @@ public class TestOptimizingQueue extends AMSTableTestBase {
queue.dispose();
}
+ @Test
+ public void testCollectDynamicAllocationLoad() {
+ DefaultTableRuntime tableRuntime = initTableWithFiles();
+ OptimizingQueue queue = buildOptimizingGroupService(tableRuntime);
+
+ // Before any poll nothing has been planned: the PENDING table is the only
demand signal —
+ // exactly what dynamic allocation must observe on a cold group with zero
optimizers.
+ DynamicAllocationState.GroupLoad before =
queue.collectDynamicAllocationLoad();
+ Assert.assertEquals(0, before.getBusyThreads());
+ Assert.assertEquals(0, before.getServiceablePlanned());
+ Assert.assertEquals(1, before.getPendingTables());
+
+ // A poll drives planning and takes the produced task: the thread is busy
from SCHEDULED
+ // (not only from ACKED), and the table is no longer PENDING.
+ TaskRuntime<?> task = queue.pollTask(optimizerThread, MAX_POLLING_TIME);
+ Assert.assertNotNull(task);
+ Assert.assertEquals(TaskRuntime.Status.SCHEDULED, task.getStatus());
+
+ DynamicAllocationState.GroupLoad after =
queue.collectDynamicAllocationLoad();
+ Assert.assertEquals(1, after.getBusyThreads());
+ Assert.assertEquals(0, after.getServiceablePlanned());
+ Assert.assertEquals(0, after.getPendingTables());
+ queue.dispose();
+ }
+
@Test
public void testPollTaskWithOverQuotaDisabled() {
DefaultTableRuntime tableRuntime = initTableWithPartitionedFiles();
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
new file mode 100644
index 000000000..d5b08ec56
--- /dev/null
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestComputeScaleUp.java
@@ -0,0 +1,231 @@
+/*
+ * 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.HashMap;
+import java.util.Map;
+
+/**
+ * Tests for {@link DynamicAllocationState#computeScaleUp}. Time is injected,
so every scenario is
+ * deterministic; the returned value is the number of
executor-parallelism-thread instances to
+ * create in this round.
+ */
+public class TestComputeScaleUp {
+
+ private static final long T0 = 0L;
+ private static final long BACKLOG_MS = 60_000L; // scheduler-backlog-timeout
default 1min
+ private static final long SUSTAINED_MS = 30_000L; //
sustained-backlog-timeout default 30s
+
+ private DynamicAllocationConfig config(int minParallelism, int
maxParallelism, int k) {
+ 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,
String.valueOf(maxParallelism));
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_PARALLELISM,
String.valueOf(k));
+ return DynamicAllocationConfig.parse(
+ new ResourceGroup.Builder("group1",
"flink").addProperties(props).build());
+ }
+
+ // --- floor enforcement: immediate, no timing gate ---
+
+ @Test
+ void floorDeficitScalesImmediately() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ // min=5, K=2: ceil(5/2) = 3 instances, on the very first evaluation.
+ Assertions.assertEquals(3, state.computeScaleUp(0, 0, 0, 0, config(5, 100,
2), T0));
+ }
+
+ @Test
+ void floorNeverExceedsMaxParallelism() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ // min=5, max=6, K=4: ceil(5/4)=2 instances would be 8 threads > max; cap
allows only 1.
+ // validate() rejects this combination up front (unreachable floor); the
cap clamp here is
+ // defense in depth for a config that bypassed validation (e.g. persisted
before upgrade).
+ Assertions.assertEquals(1, state.computeScaleUp(0, 0, 0, 0, config(5, 6,
4), T0));
+ }
+
+ // --- no demand ---
+
+ @Test
+ void noDemandReturnsZero() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ Assertions.assertEquals(0, state.computeScaleUp(4, 2, 0, 0, config(0, 100,
2), T0));
+ }
+
+ @Test
+ void idleCapacityCoveringBacklogReturnsZero() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ // busy(2) + serviceable(3) <= effective(8): idle threads will pick the
tasks up.
+ Assertions.assertEquals(0, state.computeScaleUp(8, 2, 3, 0, config(0, 100,
2), T0));
+ }
+
+ // --- immediate demand (Layer 1): backlog timer, ramp, clamp ---
+
+ @Test
+ void immediateBacklogWaitsForSchedulerBacklogTimeout() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ DynamicAllocationConfig config = config(0, 100, 2);
+ Assertions.assertEquals(0, state.computeScaleUp(2, 2, 5, 0, config, T0));
+ Assertions.assertEquals(0, state.computeScaleUp(2, 2, 5, 0, config, T0 +
BACKLOG_MS - 1));
+ Assertions.assertEquals(1, state.computeScaleUp(2, 2, 5, 0, config, T0 +
BACKLOG_MS));
+ }
+
+ @Test
+ void exponentialRampAcrossSustainedRounds() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ DynamicAllocationConfig config = config(0, 1000, 2);
+ // A large persistent backlog; effective/busy grow by the created threads
each round.
+ Assertions.assertEquals(0, state.computeScaleUp(2, 2, 100, 0, config, T0));
+ Assertions.assertEquals(1, state.computeScaleUp(2, 2, 100, 0, config, T0 +
BACKLOG_MS));
+ Assertions.assertEquals(
+ 2, state.computeScaleUp(4, 4, 100, 0, config, T0 + BACKLOG_MS +
SUSTAINED_MS));
+ Assertions.assertEquals(
+ 4, state.computeScaleUp(8, 8, 100, 0, config, T0 + BACKLOG_MS + 2 *
SUSTAINED_MS));
+ Assertions.assertEquals(
+ 8, state.computeScaleUp(16, 16, 100, 0, config, T0 + BACKLOG_MS + 3 *
SUSTAINED_MS));
+ }
+
+ @Test
+ void sustainedCadenceGatesConsecutiveAdds() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ DynamicAllocationConfig config = config(0, 1000, 2);
+ Assertions.assertEquals(0, state.computeScaleUp(2, 2, 100, 0, config, T0));
+ Assertions.assertEquals(1, state.computeScaleUp(2, 2, 100, 0, config, T0 +
BACKLOG_MS));
+ // 10s after the first add: sustained-backlog-timeout (30s) has not
elapsed.
+ Assertions.assertEquals(
+ 0, state.computeScaleUp(4, 4, 100, 0, config, T0 + BACKLOG_MS +
10_000));
+ Assertions.assertEquals(
+ 2, state.computeScaleUp(4, 4, 100, 0, config, T0 + BACKLOG_MS +
SUSTAINED_MS));
+ }
+
+ @Test
+ void rampResetsWhenClampBinds() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ DynamicAllocationConfig config = config(0, 1000, 1);
+ long t1 = T0 + BACKLOG_MS;
+ long t2 = t1 + SUSTAINED_MS;
+ long t3 = t2 + SUSTAINED_MS;
+ long t4 = t3 + SUSTAINED_MS;
+ Assertions.assertEquals(0, state.computeScaleUp(2, 2, 10, 0, config, T0));
+ // A: want=10 > ramp=1 -> add 1, ramp doubles to 2.
+ Assertions.assertEquals(1, state.computeScaleUp(2, 2, 10, 0, config, t1));
+ // B: want=1 < ramp=2 -> clamp binds: add 1 and the ramp resets to 1
(Spark semantics),
+ // instead of keeping a grown ramp no demand justified.
+ Assertions.assertEquals(1, state.computeScaleUp(3, 3, 1, 0, config, t2));
+ // C: demand returns: ramp restarts from 1, not from the stale doubled
value.
+ Assertions.assertEquals(1, state.computeScaleUp(4, 4, 10, 0, config, t3));
+ // D: ramp doubling resumes normally.
+ Assertions.assertEquals(2, state.computeScaleUp(5, 5, 10, 0, config, t4));
+ }
+
+ @Test
+ void addIsCappedByMaxParallelismAndStopsAtCap() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ DynamicAllocationConfig config = config(0, 10, 2);
+ Assertions.assertEquals(0, state.computeScaleUp(8, 8, 100, 0, config, T0));
+ // Only floor((10-8)/2) = 1 instance fits under the cap.
+ Assertions.assertEquals(1, state.computeScaleUp(8, 8, 100, 0, config, T0 +
BACKLOG_MS));
+ // At the cap: nothing more can be created no matter the backlog.
+ Assertions.assertEquals(
+ 0, state.computeScaleUp(10, 10, 100, 0, config, T0 + BACKLOG_MS +
SUSTAINED_MS));
+ }
+
+ @Test
+ void floorRoundsResetDemandTiming() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ DynamicAllocationConfig config = config(2, 100, 1);
+ // A demand phase establishes a passed cadence gate.
+ Assertions.assertEquals(0, state.computeScaleUp(2, 2, 10, 0, config, T0));
+ Assertions.assertEquals(1, state.computeScaleUp(2, 2, 10, 0, config, T0 +
BACKLOG_MS));
+ // Optimizers die: effective drops below the floor; floor rounds bypass
demand timing.
+ Assertions.assertEquals(2, state.computeScaleUp(0, 0, 0, 0, config, T0 +
BACKLOG_MS + 10_000));
+ // After recovery, fresh demand must re-prove backlog persistence instead
of firing
+ // immediately through the stale gate left over from the earlier demand
phase.
+ long recovered = T0 + BACKLOG_MS + 120_000;
+ Assertions.assertEquals(0, state.computeScaleUp(2, 2, 10, 0, config,
recovered));
+ Assertions.assertEquals(1, state.computeScaleUp(2, 2, 10, 0, config,
recovered + BACKLOG_MS));
+ }
+
+ @Test
+ void oscillatingDemandResetsBacklogTimer() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ DynamicAllocationConfig config = config(0, 100, 2);
+ // A trickle whose tasks are drained between rounds must not accumulate
toward the timeout:
+ // arrival keeping pace with processing is not under-capacity.
+ Assertions.assertEquals(0, state.computeScaleUp(2, 2, 5, 0, config, T0));
+ Assertions.assertEquals(0, state.computeScaleUp(2, 2, 0, 0, config, T0 +
30_000)); // resets
+ Assertions.assertEquals(0, state.computeScaleUp(2, 2, 5, 0, config, T0 +
45_000)); // restarts
+ Assertions.assertEquals(
+ 0, state.computeScaleUp(2, 2, 5, 0, config, T0 + 45_000 + BACKLOG_MS -
1));
+ Assertions.assertEquals(1, state.computeScaleUp(2, 2, 5, 0, config, T0 +
45_000 + BACKLOG_MS));
+ }
+
+ // --- future demand (Layer 2): pending tables, all threads busy, one
instance ---
+
+ @Test
+ void coldStartWithPendingTablesAddsOneInstance() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ DynamicAllocationConfig config = config(0, 100, 4);
+ // Zero optimizers: no pollTask, no planning, no PLANNED tasks — only
PENDING tables are
+ // observable. Layer 2 must ignite the group.
+ Assertions.assertEquals(0, state.computeScaleUp(0, 0, 0, 3, config, T0));
+ Assertions.assertEquals(1, state.computeScaleUp(0, 0, 0, 3, config, T0 +
BACKLOG_MS));
+ }
+
+ @Test
+ void futureDemandRequiresAllThreadsBusy() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ DynamicAllocationConfig config = config(0, 100, 2);
+ // An idle thread exists: it will drive planning via pollTask, so pending
tables alone are
+ // not a scale signal.
+ Assertions.assertEquals(0, state.computeScaleUp(4, 3, 0, 5, config, T0));
+ Assertions.assertEquals(0, state.computeScaleUp(4, 3, 0, 5, config, T0 +
BACKLOG_MS));
+ }
+
+ @Test
+ void futureDemandAddsOneInstanceRegardlessOfPendingTableCount() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ DynamicAllocationConfig config = config(0, 100, 4);
+ // Pending tables are not quantified demand (tasks per table are unknown
before planning);
+ // Layer 2 probes with a single instance and lets Layer 1 take over once
tasks materialize.
+ Assertions.assertEquals(0, state.computeScaleUp(0, 0, 0, 100, config, T0));
+ Assertions.assertEquals(1, state.computeScaleUp(0, 0, 0, 100, config, T0 +
BACKLOG_MS));
+ }
+
+ @Test
+ void futureDemandDoesNotRamp() {
+ DynamicAllocationState state = new DynamicAllocationState();
+ DynamicAllocationConfig config = config(0, 100, 4);
+ Assertions.assertEquals(0, state.computeScaleUp(0, 0, 0, 10, config, T0));
+ Assertions.assertEquals(1, state.computeScaleUp(0, 0, 0, 10, config, T0 +
BACKLOG_MS));
+ // Still pending-only demand after the first instance registered and is
busy: another single
+ // instance, not an exponentially grown batch.
+ Assertions.assertEquals(
+ 1, state.computeScaleUp(4, 4, 0, 10, config, T0 + BACKLOG_MS +
SUSTAINED_MS));
+ Assertions.assertEquals(
+ 1, state.computeScaleUp(8, 8, 0, 10, config, T0 + BACKLOG_MS + 2 *
SUSTAINED_MS));
+ }
+}
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 837584285..66d5aa41f 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
@@ -70,6 +70,70 @@ public class TestDynamicAllocationConfig {
Assertions.assertThrows(IllegalArgumentException.class, () ->
parseAndValidate(group(props)));
}
+ @Test
+ void executorParallelismDefaultsToOne() {
+ DynamicAllocationConfig config =
DynamicAllocationConfig.parse(group(enabledProps()));
+ assertDoesNotThrow(config::validate);
+ Assertions.assertEquals(1, config.getExecutorParallelism());
+ }
+
+ @Test
+ void executorParallelismIsParsed() {
+ Map<String, String> props = enabledProps();
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_PARALLELISM,
"4");
+ DynamicAllocationConfig config =
DynamicAllocationConfig.parse(group(props));
+ assertDoesNotThrow(config::validate);
+ Assertions.assertEquals(4, config.getExecutorParallelism());
+ }
+
+ @Test
+ void executorParallelismBelowOneIsRejected() {
+ Map<String, String> props = enabledProps();
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_PARALLELISM,
"0");
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
parseAndValidate(group(props)));
+ }
+
+ @Test
+ void executorParallelismAboveMaxParallelismIsRejected() {
+ // A single K-thread instance would already exceed the max-parallelism
cap, so scale-up
+ // could never create anything; reject up front instead of leaving a
silent no-op group.
+ Map<String, String> props = enabledProps();
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_PARALLELISM,
"32");
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
parseAndValidate(group(props)));
+ }
+
+ @Test
+ void executorParallelismMakingFloorUnreachableIsRejected() {
+ // min=5, max=6, K=4: covering the floor needs ceil(5/4)=2 instances = 8
threads > max,
+ // so the floor could never be satisfied in K units and the group would
silently sit
+ // below its floor forever.
+ Map<String, String> props = enabledProps();
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM, "5");
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM, "6");
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_PARALLELISM,
"4");
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
parseAndValidate(group(props)));
+ }
+
+ @Test
+ void reachableFloorInExecutorParallelismUnitsIsAccepted() {
+ // min=5, max=8, K=4: ceil(5/4)=2 instances = 8 threads fits under max.
+ Map<String, String> props = enabledProps();
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM, "5");
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM, "8");
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_PARALLELISM,
"4");
+ assertDoesNotThrow(() -> parseAndValidate(group(props)));
+ }
+
+ @Test
+ void malformedExecutorParallelismIsRejectedAtParse() {
+ // Same parse() contract as min/max-parallelism: a malformed numeric is
rejected at parse
+ // regardless of enabled.
+ Map<String, String> props = new HashMap<>();
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_PARALLELISM,
"abc");
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
DynamicAllocationConfig.parse(group(props)));
+ }
+
@Test
void enabledWithUnparsableMinParallelismIsRejected() {
// resolveMinParallelism() is lenient (legacy/keeper path), but an
opted-in group must not
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestDynamicAllocationState.java
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestDynamicAllocationState.java
new file mode 100644
index 000000000..fc5233aee
--- /dev/null
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestDynamicAllocationState.java
@@ -0,0 +1,175 @@
+/*
+ * 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.server.optimizing.TaskRuntime;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+public class TestDynamicAllocationState {
+
+ private DynamicAllocationState.TableDemand demand(
+ int plannedCount, double targetQuota, int occupiedThreads) {
+ return new DynamicAllocationState.TableDemand(plannedCount, targetQuota,
occupiedThreads);
+ }
+
+ // --- serviceablePlannedCount: proportional mode (targetQuota <= 1) ---
+ // The per-table quota limit is ceil(targetQuota * availableCore), so adding
threads raises the
+ // limit; a quota-blocked backlog is still serviceable demand and counts in
full.
+
+ @Test
+ void proportionalQuotaCountsAllPlannedTasks() {
+ Assertions.assertEquals(
+ 10,
+ DynamicAllocationState.serviceablePlannedCount(
+ Collections.singletonList(demand(10, 0.5, 4))));
+ }
+
+ @Test
+ void targetQuotaOfExactlyOneIsProportional() {
+ // Absolute mode starts strictly above 1 (see OptimizingUtil quota
resolution).
+ Assertions.assertEquals(
+ 7,
+ DynamicAllocationState.serviceablePlannedCount(
+ Collections.singletonList(demand(7, 1.0, 3))));
+ }
+
+ // --- serviceablePlannedCount: absolute mode (targetQuota > 1) ---
+ // The limit is a fixed thread count that scaling cannot raise; only free
slots are serviceable.
+
+ @Test
+ void absoluteQuotaCountsOnlyFreeSlots() {
+ Assertions.assertEquals(
+ 2,
+ DynamicAllocationState.serviceablePlannedCount(
+ Collections.singletonList(demand(10, 3.0, 1))));
+ }
+
+ @Test
+ void absoluteQuotaExhaustedCountsZero() {
+ Assertions.assertEquals(
+ 0,
+ DynamicAllocationState.serviceablePlannedCount(
+ Collections.singletonList(demand(5, 2.0, 2))));
+ }
+
+ @Test
+ void absoluteQuotaOverOccupiedDoesNotGoNegative() {
+ // occupied can transiently exceed the limit (e.g. after a quota config
decrease); the table
+ // must contribute zero, not a negative count offsetting other tables.
+ Assertions.assertEquals(
+ 0,
+ DynamicAllocationState.serviceablePlannedCount(
+ Collections.singletonList(demand(5, 2.0, 3))));
+ }
+
+ @Test
+ void absoluteQuotaFractionalLimitTruncates() {
+ // The limit is (int) targetQuota, matching the poll gate's getQuotaLimit
cast: 2.5 -> 2
+ // slots, so 2 occupied threads leave zero free.
+ Assertions.assertEquals(
+ 0,
+ DynamicAllocationState.serviceablePlannedCount(
+ Collections.singletonList(demand(5, 2.5, 2))));
+ }
+
+ @Test
+ void absoluteQuotaFreeSlotsCappedByPlanned() {
+ // Free slots exceed the planned backlog; only actual tasks count.
+ Assertions.assertEquals(
+ 3,
+ DynamicAllocationState.serviceablePlannedCount(
+ Collections.singletonList(demand(3, 8.0, 1))));
+ }
+
+ // --- serviceablePlannedCount: aggregation ---
+
+ @Test
+ void mixedModesSumPerTable() {
+ Assertions.assertEquals(
+ 12,
+ DynamicAllocationState.serviceablePlannedCount(
+ Arrays.asList(demand(10, 0.5, 4), demand(10, 3.0, 1))));
+ }
+
+ @Test
+ void emptyDemandsCountZero() {
+ Assertions.assertEquals(
+ 0,
DynamicAllocationState.serviceablePlannedCount(Collections.emptyList()));
+ }
+
+ // --- isPlanningBound ---
+ // Idle threads while tables wait as PENDING and no PLANNED tasks
materialize means the
+ // bottleneck is planning (maxPlanningParallelism), not thread capacity:
scaling out would only
+ // add more idle threads. The condition is surfaced as a warning, never as a
scale-out.
+
+ @Test
+ void planningBoundWhenThreadsIdleTablesPendingAndNothingPlanned() {
+ Assertions.assertTrue(DynamicAllocationState.isPlanningBound(4, 2, 0, 10));
+ }
+
+ @Test
+ void notPlanningBoundWhenPlannedTasksAwaitPickup() {
+ // Idle threads have work to poll; planning is keeping up.
+ Assertions.assertFalse(DynamicAllocationState.isPlanningBound(4, 2, 3,
10));
+ }
+
+ @Test
+ void notPlanningBoundWhenAllThreadsBusy() {
+ // Saturated threads are Layer-2 territory, not a planning bottleneck
signal.
+ Assertions.assertFalse(DynamicAllocationState.isPlanningBound(4, 4, 0,
10));
+ }
+
+ @Test
+ void notPlanningBoundWithoutPendingTables() {
+ Assertions.assertFalse(DynamicAllocationState.isPlanningBound(4, 2, 0, 0));
+ }
+
+ @Test
+ void notPlanningBoundOnColdGroup() {
+ // Zero optimizers is the cold-start case handled by future demand, not a
planning issue.
+ Assertions.assertFalse(DynamicAllocationState.isPlanningBound(0, 0, 0,
10));
+ }
+
+ // --- occupiesThread ---
+ // A task occupies an optimizer thread from the moment it is assigned
(SCHEDULED, set by
+ // pollTask) until it terminates; counting only ACKED would overestimate
headroom during the
+ // poll-to-ack window.
+
+ @Test
+ void scheduledTaskOccupiesAThread() {
+
Assertions.assertTrue(DynamicAllocationState.occupiesThread(TaskRuntime.Status.SCHEDULED));
+ }
+
+ @Test
+ void ackedTaskOccupiesAThread() {
+
Assertions.assertTrue(DynamicAllocationState.occupiesThread(TaskRuntime.Status.ACKED));
+ }
+
+ @Test
+ void terminalAndQueuedStatusesDoNotOccupyThreads() {
+
Assertions.assertFalse(DynamicAllocationState.occupiesThread(TaskRuntime.Status.PLANNED));
+
Assertions.assertFalse(DynamicAllocationState.occupiesThread(TaskRuntime.Status.SUCCESS));
+
Assertions.assertFalse(DynamicAllocationState.occupiesThread(TaskRuntime.Status.FAILED));
+
Assertions.assertFalse(DynamicAllocationState.occupiesThread(TaskRuntime.Status.CANCELED));
+ }
+}
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestPendingRegistrations.java
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestPendingRegistrations.java
new file mode 100644
index 000000000..2ec07734e
--- /dev/null
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestPendingRegistrations.java
@@ -0,0 +1,87 @@
+/*
+ * 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.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for {@link PendingRegistrations}: requested-but-not-yet-registered
optimizer capacity.
+ * Registration is optimizer-driven (the pod self-registers after boot), so a
request that never
+ * registers — image pull failure, exhausted ResourceQuota, crash loop — must
not count as capacity
+ * forever: it would freeze scale-up below the real demand. Heartbeat expiry
cannot cover this
+ * window because it only starts after registration.
+ */
+public class TestPendingRegistrations {
+
+ private static final long TIMEOUT_MS = 180_000L; // boot timeout, must
exceed pod boot time
+ private static final long T0 = 0L;
+
+ @Test
+ void requestedThreadsArePending() {
+ PendingRegistrations pending = new PendingRegistrations(TIMEOUT_MS);
+ pending.requested("r1", 4, T0);
+ pending.requested("r2", 4, T0);
+ Assertions.assertEquals(8, pending.pendingThreads(T0 + 1_000));
+ }
+
+ @Test
+ void registrationRemovesThePendingEntry() {
+ PendingRegistrations pending = new PendingRegistrations(TIMEOUT_MS);
+ pending.requested("r1", 4, T0);
+ pending.requested("r2", 4, T0);
+ pending.registered("r1");
+ Assertions.assertEquals(4, pending.pendingThreads(T0 + 1_000));
+ }
+
+ @Test
+ void requestFailureRemovesThePendingEntryImmediately() {
+ PendingRegistrations pending = new PendingRegistrations(TIMEOUT_MS);
+ pending.requested("r1", 4, T0);
+ pending.failed("r1");
+ Assertions.assertEquals(0, pending.pendingThreads(T0 + 1_000));
+ }
+
+ @Test
+ void entriesExpireAfterTheBootTimeout() {
+ PendingRegistrations pending = new PendingRegistrations(TIMEOUT_MS);
+ pending.requested("r1", 4, T0);
+ // Within the timeout the phantom capacity is accepted: evicting a
legitimately booting pod
+ // would cause duplicate scale-ups, which is worse than a few conservative
rounds.
+ Assertions.assertEquals(4, pending.pendingThreads(T0 + TIMEOUT_MS - 1));
+ Assertions.assertEquals(0, pending.pendingThreads(T0 + TIMEOUT_MS));
+ }
+
+ @Test
+ void expiryIsPerEntry() {
+ PendingRegistrations pending = new PendingRegistrations(TIMEOUT_MS);
+ pending.requested("r1", 4, T0);
+ pending.requested("r2", 2, T0 + 60_000);
+ Assertions.assertEquals(2, pending.pendingThreads(T0 + TIMEOUT_MS));
+ }
+
+ @Test
+ void unknownResourceIdIsIgnored() {
+ PendingRegistrations pending = new PendingRegistrations(TIMEOUT_MS);
+ pending.requested("r1", 4, T0);
+ pending.registered("unknown");
+ pending.failed("also-unknown");
+ Assertions.assertEquals(4, pending.pendingThreads(T0 + 1_000));
+ }
+}
diff --git
a/amoro-common/src/main/java/org/apache/amoro/OptimizerProperties.java
b/amoro-common/src/main/java/org/apache/amoro/OptimizerProperties.java
index c69a8aa94..134811580 100644
--- a/amoro-common/src/main/java/org/apache/amoro/OptimizerProperties.java
+++ b/amoro-common/src/main/java/org/apache/amoro/OptimizerProperties.java
@@ -78,6 +78,12 @@ public class OptimizerProperties {
public static final int DYNAMIC_ALLOCATION_MAX_PARALLELISM_LIMIT = 1024;
+ /** @since 0.9.0 */
+ public static final String DYNAMIC_ALLOCATION_EXECUTOR_PARALLELISM =
+ "dynamic-allocation.executor-parallelism";
+
+ public static final int DYNAMIC_ALLOCATION_EXECUTOR_PARALLELISM_DEFAULT = 1;
+
/** @since 0.9.0 */
public static final String DYNAMIC_ALLOCATION_SCHEDULER_BACKLOG_TIMEOUT =
"dynamic-allocation.scheduler-backlog-timeout";
diff --git a/docs/admin-guides/managing-optimizers.md
b/docs/admin-guides/managing-optimizers.md
index 35a0d935f..5abc4faf1 100644
--- a/docs/admin-guides/managing-optimizers.md
+++ b/docs/admin-guides/managing-optimizers.md
@@ -288,8 +288,17 @@ The optimizer group supports the following properties:
| cache-max-total-size | All | No | 128mb
| Max
total size in optimier cache.
[...]
| cache-max-entry-size | All | No | 64mb
| Max
entry size in optimizer cache.
[...]
| cache-timeout | All | No | 10min
| Timeout
in optimizer cache.
[...]
-| min-parallelism | All | No | 0
| The
minimum total parallelism (CPU cores) that the optimizer group should maintain.
When the total cores of running optimizers fall below this value,
`OptimizerGroupKeeper` will automatically scale out new optimizers. Set to `0`
to disable auto-scaling. Note: The behavior of the auto-scaling mechanism is
controlled by the AMS-level configu [...]
+| min-parallelism | All | No | 0
|
Deprecated since 0.9.0 in favor of `dynamic-allocation.min-parallelism`; still
honored as a fallback. The minimum total parallelism (CPU cores) that the
optimizer group should maintain. When the total cores of running optimizers
fall below this value, `OptimizerGroupKeeper` will automatically scale out new
optimizers. Set to `0` to disable [...]
| shutdown-timeout-ms | All | No | 600000(10min)
|
Graceful shutdown timeout in milliseconds. On shutdown the optimizer waits up
to this long for in-progress tasks to complete before force-interrupting them.
For Kubernetes optimizers, the pod's `terminationGracePeriodSeconds` is derived
from this value plus a 30s buffer. For Flink optimizers, the effective wait is
additionally capped below [...]
+| dynamic-allocation.enabled | All | No | false
| Whether
to enable dynamic resource allocation (AIP-5) for this group: the group then
grows its optimizer count automatically in response to optimizing demand,
bounded by `dynamic-allocation.max-parallelism`. Not supported on
externally-registered optimizers. Manual optimizer operations
(scale-out/release via the dashboard) are not recommen [...]
+| dynamic-allocation.min-parallelism | All | No | 0
| Lower
bound on the group's total optimizer threads under dynamic allocation.
Supersedes the deprecated flat `min-parallelism`.
[...]
+| dynamic-allocation.max-parallelism | All | Yes (when enabled) | N/A
| Upper
bound on the group's total optimizer threads under dynamic allocation; must not
exceed 1024. Also configure Kubernetes `ResourceQuota`/`LimitRange` as the
authoritative cluster-side limit.
[...]
+| 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.
[...]
| 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.
[...]
