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 52345d2ccb..6831d57c79 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.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.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 @@ private void loadOptimizingQueues(List tableRuntimeList) { 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 @@ private void registerOptimizer(OptimizerInstance optimizer, boolean needPersiste 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 void createResourceGroup(ResourceGroup resourceGroup) { 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 void dispose() { optimizingQueueByGroup.values().forEach(OptimizingQueue::dispose); optimizerKeeper.dispose(); optimizerGroupKeeper.dispose(); + optimizerScaleKeeper.dispose(); tableHandlerChain.dispose(); optimizingQueueByGroup.clear(); optimizingQueueByToken.clear(); @@ -497,6 +513,7 @@ protected void initHandler(List tableRuntimeList) { .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 @@ protected void processTask(OptimizerGroupKeepingTask keepingTask) { 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 @@ protected void processTask(OptimizerGroupKeepingTask keepingTask) { 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 @@ protected void processTask(OptimizerGroupKeepingTask keepingTask) { 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 { + + // 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 scaleStates = new ConcurrentHashMap<>(); + private final Map pendingRegistrations = + new ConcurrentHashMap<>(); + private final Set watchedGroups = ConcurrentHashMap.newKeySet(); + private final Map 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 52eb46b4cc..bbbd6b288e 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.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 List> collectTasks(Predicate> predicate) { .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 plannedByTable = Maps.newHashMap(); + Map 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 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 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) 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 0759f1e367..8bb2d0eb1a 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.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 void removeTable(DefaultTableRuntime tableRuntime) { Map 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 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 ad62ad291a..15c9de8417 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 @@ private DynamicAllocationConfig( boolean enabled, Integer minParallelism, Integer maxParallelism, + int executorParallelism, Duration schedulerBacklogTimeout, Duration sustainedBacklogTimeout, Duration executorIdleTimeout, @@ -64,6 +66,7 @@ private 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 static DynamicAllocationConfig parse(ResourceGroup group) { 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 void validate() { 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 int getMaxParallelism() { 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 0000000000..9ac84334a3 --- /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. + * + *

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. + * + *

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 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 0000000000..abc80c537d --- /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 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 0000000000..8a57acb6c4 --- /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 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 initializedField = + DynFields.builder().hiddenImpl(Containers.class, "isInitialized").build(); + initializedField.asStatic().set(false); + } + } + + private void setupMockContainer(Supplier targetGroupNameSupplier) throws Exception { + TestOptimizerGroupKeeper.MockOptimizerContainer mockContainer = + new TestOptimizerGroupKeeper.MockOptimizerContainer( + resourceAvailable, + scaleOutCallCount, + registerInfo -> { + Function registrar = + optimizerRegistrar; + return registrar == null ? null : registrar.apply(registerInfo); + }, + targetGroupNameSupplier); + + DynFields.UnboundField initializedField = + DynFields.builder().hiddenImpl(Containers.class, "isInitialized").build(); + if (!initializedField.asStatic().get()) { + originIsInitialized = false; + initializedField.asStatic().set(true); + } + + DynFields.UnboundField> containersField = + DynFields.builder().hiddenImpl(Containers.class, "globalContainers").build(); + Map globalContainers = containersField.asStatic().get(); + + ContainerMetadata metadata = + new ContainerMetadata( + MOCK_CONTAINER_NAME, TestOptimizerGroupKeeper.MockOptimizerContainer.class.getName()); + Map 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 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 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 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 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 772f15c01f..78a9e22e3c 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.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 void testPollTask() { 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 0000000000..d5b08ec566 --- /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 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 8375842852..66d5aa41f4 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 @@ void maxParallelismAboveHardLimitIsRejected() { 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 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 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 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 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 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 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 0000000000..fc5233aee8 --- /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 0000000000..2ec07734ef --- /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 c69a8aa94b..1348115802 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 35a0d935fc..5abc4faf15 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 configurations `optimizer-group.min-parallelism-check-interval` and `optimizer-group.max-keeping-attempts`.| +| 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 auto-scaling. Note: The behavior of the auto-scaling mechanism is controlled by the AMS-level configurations `optimizer-group.min-parallelism-check-interval` and `optimizer-group.max-keeping-attempts`. For groups with dynamic allocation enabled, the floor is maintained by the dynamic allocation scale keeper instead. | | 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 `task.cancellation.timeout`. | +| 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 recommended on groups with dynamic allocation enabled. | +| 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.\ | Flink | No | N/A | Any flink config options could be overwritten, priority is optimizing-group > optimizing-container > flink-conf.yaml. | | spark-conf.\ | Spark | No | N/A | Any spark config options could be overwritten, priority is optimizing-group > optimizing-container > spark-defaults.conf. |