This is an automated email from the ASF dual-hosted git repository.
czy006 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 8b03fec11 [AMORO-4253] AIP-5 Phase 1: dynamic allocation configuration
and validation (#4254)
8b03fec11 is described below
commit 8b03fec1105aa4b29b5f4d09815c62628187c115
Author: Jiwon Park <[email protected]>
AuthorDate: Fri Jul 10 12:22:05 2026 +0900
[AMORO-4253] AIP-5 Phase 1: dynamic allocation configuration and validation
(#4254)
* [AMORO-4253] AIP-5 Phase 1: dynamic allocation configuration and
validation
First implementation phase of AIP-5 (Dynamic Resource Allocation for
Optimizer, #4191). Configuration foundation only; no runtime scaling
behavior. Dynamic allocation is opt-in and disabled by default, so
existing groups are unaffected.
- OptimizerProperties: declare the dynamic-allocation.* constants (with
@since); deprecate the flat min-parallelism in favor of the namespaced
dynamic-allocation.min-parallelism, still honored as a fallback.
- DynamicAllocationConfig: parse(ResourceGroup) + validate() enforcing
the AIP-5 constraints only when enabled (max-parallelism mandatory,
min <= max <= 1024, executor-idle-timeout >= 30s, positive durations,
reject the external container). resolveMinParallelism centralizes the
namespaced -> legacy -> 0 resolution.
- OptimizerGroupController: reject an invalid DRA config with HTTP 400 on
create/update before persisting.
- DefaultOptimizingService: startup load is fail-safe (invalid config
logs a warning and falls back to DRA-disabled instead of crashing AMS);
the keeper min-parallelism auto-reset is skipped when DRA is
effectively enabled, and otherwise writes the key the group actually
reads (fixing an endless no-op loop for namespaced groups).
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4253] Strictly validate min-parallelism on opted-in DRA groups
resolveMinParallelism() is intentionally lenient (unparsable -> 0) for the
keeper hot path and legacy compatibility, but validate() reused that
leniency
for opted-in groups: an enabled group with
'dynamic-allocation.min-parallelism'
set to a non-numeric value silently degraded to 0, and a negative value
passed
the max >= min check unchecked. This was asymmetric with max-parallelism,
which
is parsed strictly and rejected with HTTP 400.
Retain the raw min-parallelism string at parse time so validate() can tell
"explicitly configured but unparsable" from "unset" (the resolved int
collapses
both to 0). When DRA is enabled, an explicitly configured min-parallelism
that
is unparsable or negative is now rejected. The keeper hot path stays
lenient.
Also document the getMaxParallelism() unboxing precondition (nullable until
validated) and leave a TODO for realistic lower bounds on the polling
timeouts.
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4253] Unify min-parallelism with max-parallelism as nullable
Integer
Drop the dual int/String representation of min-parallelism in favor of a
single nullable Integer, mirroring max-parallelism: the value is strictly
parsed in parse() (a malformed numeric throws there, honoring parse()'s
contract, instead of being silently degraded to 0 by the lenient keeper
resolver), and null encodes "unset". validate() then only checks the
semantic bounds (non-negative, max >= min).
Also align the trim semantics of the lenient resolveMinParallelism() path
with the strict one so a whitespace-padded value (e.g. " 5 ") that
validate() accepts resolves to the same floor the keeper reads, instead of
silently degrading to 0 at runtime.
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4253] Cite the actually-used key in the min-parallelism parse error
When min-parallelism is configured only via the deprecated flat key, a
malformed value reported the namespaced 'dynamic-allocation.min-parallelism'
the user never set. Pick the value and its source key together so the error
names the key the value actually came from, folding the former
rawMinParallelism helper into parseMinParallelism.
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4253] Drop the inert DRA validation block from queue loading
The startup validate/fallback block in loadOptimizingQueues was
behaviorally inert: the disable is already implicit (isEffectivelyEnabled()
treats an invalid config as disabled wherever DRA is consumed), and in
Phase 1 nothing strictly parses DRA config at load, so there was no crash
to guard against. Validating inside the queue loader also mixed an unrelated
concern into it. Surfacing invalid configs as an alertable signal belongs
with the metric that lands in the observability phase.
Signed-off-by: Jiwon Park <[email protected]>
---------
Signed-off-by: Jiwon Park <[email protected]>
---
.../amoro/server/DefaultOptimizingService.java | 28 +-
.../controller/OptimizerGroupController.java | 18 +-
.../optimizing/dra/DynamicAllocationConfig.java | 362 +++++++++++++++++++++
.../amoro/server/TestOptimizerGroupController.java | 44 +++
.../amoro/server/TestOptimizerGroupKeeper.java | 61 ++++
.../dra/TestDynamicAllocationConfig.java | 245 ++++++++++++++
.../java/org/apache/amoro/OptimizerProperties.java | 58 +++-
7 files changed, 799 insertions(+), 17 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 efc1c375b..52345d2cc 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
@@ -45,6 +45,7 @@ import org.apache.amoro.server.optimizing.OptimizingProcess;
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.persistence.StatedPersistentBase;
import org.apache.amoro.server.persistence.mapper.OptimizerMapper;
import org.apache.amoro.server.persistence.mapper.ResourceMapper;
@@ -840,19 +841,7 @@ public class DefaultOptimizingService extends
StatedPersistentBase
}
public int getMinParallelism(ResourceGroup resourceGroup) {
- if (!resourceGroup
- .getProperties()
- .containsKey(OptimizerProperties.OPTIMIZER_GROUP_MIN_PARALLELISM)) {
- return 0;
- }
- String minParallelism =
-
resourceGroup.getProperties().get(OptimizerProperties.OPTIMIZER_GROUP_MIN_PARALLELISM);
- try {
- return Integer.parseInt(minParallelism);
- } catch (Throwable t) {
- LOG.warn("Illegal minParallelism : {}, will use default value 0",
minParallelism, t);
- return 0;
- }
+ return DynamicAllocationConfig.resolveMinParallelism(resourceGroup);
}
public int tryKeeping(ResourceGroup resourceGroup) {
@@ -921,6 +910,17 @@ 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(),
@@ -930,7 +930,7 @@ public class DefaultOptimizingService extends
StatedPersistentBase
resourceGroup
.getProperties()
.put(
- OptimizerProperties.OPTIMIZER_GROUP_MIN_PARALLELISM,
+
DynamicAllocationConfig.effectiveMinParallelismKey(resourceGroup),
String.valueOf(minParallelism - requiredCores));
updateResourceGroup(resourceGroup);
optimizerManager.updateResourceGroup(resourceGroup);
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/OptimizerGroupController.java
b/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/OptimizerGroupController.java
index fac5b8967..2362018b6 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/OptimizerGroupController.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/OptimizerGroupController.java
@@ -31,6 +31,7 @@ import org.apache.amoro.server.dashboard.response.PageResult;
import org.apache.amoro.server.dashboard.utils.PropertiesUtil;
import org.apache.amoro.server.manager.AbstractOptimizerContainer;
import org.apache.amoro.server.optimizing.OptimizingStatus;
+import org.apache.amoro.server.optimizing.dra.DynamicAllocationConfig;
import org.apache.amoro.server.resource.ContainerMetadata;
import org.apache.amoro.server.resource.Containers;
import org.apache.amoro.server.resource.OptimizerInstance;
@@ -242,7 +243,9 @@ public class OptimizerGroupController {
validateGroupName(name);
ResourceGroup.Builder builder = new ResourceGroup.Builder(name, container);
builder.addProperties(properties);
- optimizerManager.createResourceGroup(builder.build());
+ ResourceGroup resourceGroup = builder.build();
+ validateDynamicAllocation(resourceGroup);
+ optimizerManager.createResourceGroup(resourceGroup);
ctx.json(OkResponse.of("The optimizer group has been successfully
created."));
}
@@ -257,7 +260,9 @@ public class OptimizerGroupController {
Map<String, String> properties = PropertiesUtil.sanitizeProperties((Map)
map.get("properties"));
ResourceGroup.Builder builder = new ResourceGroup.Builder(name, container);
builder.addProperties(properties);
- optimizerManager.updateResourceGroup(builder.build());
+ ResourceGroup resourceGroup = builder.build();
+ validateDynamicAllocation(resourceGroup);
+ optimizerManager.updateResourceGroup(resourceGroup);
ctx.json(OkResponse.of("The optimizer group has been successfully
updated."));
}
@@ -283,6 +288,15 @@ public class OptimizerGroupController {
.collect(Collectors.toList())));
}
+ private void validateDynamicAllocation(ResourceGroup resourceGroup) {
+ DynamicAllocationConfig.warnDeprecatedMinParallelism(resourceGroup);
+ try {
+ DynamicAllocationConfig.parse(resourceGroup).validate();
+ } catch (IllegalArgumentException e) {
+ throw new BadRequestException(e.getMessage());
+ }
+ }
+
private void validateGroupName(String groupName) {
if (StringUtils.isEmpty(groupName)) {
throw new BadRequestException(
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
new file mode 100644
index 000000000..ad62ad291
--- /dev/null
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationConfig.java
@@ -0,0 +1,362 @@
+/*
+ * 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.Constants;
+import org.apache.amoro.OptimizerProperties;
+import org.apache.amoro.config.ConfigHelpers;
+import org.apache.amoro.resource.ResourceGroup;
+import org.apache.amoro.utils.PropertyUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.Map;
+
+/**
+ * Dynamic resource allocation (DRA) configuration of a resource group
(AIP-5). Parsed from the
+ * group's properties; {@link #validate()} enforces the AIP-5 constraints.
+ */
+public class DynamicAllocationConfig {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(DynamicAllocationConfig.class);
+
+ private final String groupName;
+ private final String container;
+ private final boolean enabled;
+ private final Integer minParallelism;
+ private final Integer maxParallelism;
+ private final Duration schedulerBacklogTimeout;
+ private final Duration sustainedBacklogTimeout;
+ private final Duration executorIdleTimeout;
+ private final Duration scaleDownCooldown;
+ private final Duration drainTimeout;
+
+ private DynamicAllocationConfig(
+ String groupName,
+ String container,
+ boolean enabled,
+ Integer minParallelism,
+ Integer maxParallelism,
+ Duration schedulerBacklogTimeout,
+ Duration sustainedBacklogTimeout,
+ Duration executorIdleTimeout,
+ Duration scaleDownCooldown,
+ Duration drainTimeout) {
+ this.groupName = groupName;
+ this.container = container;
+ this.enabled = enabled;
+ this.minParallelism = minParallelism;
+ this.maxParallelism = maxParallelism;
+ this.schedulerBacklogTimeout = schedulerBacklogTimeout;
+ this.sustainedBacklogTimeout = sustainedBacklogTimeout;
+ this.executorIdleTimeout = executorIdleTimeout;
+ this.scaleDownCooldown = scaleDownCooldown;
+ this.drainTimeout = drainTimeout;
+ }
+
+ /**
+ * Parse the DRA configuration of a resource group. Malformed numeric or
duration values throw
+ * {@link IllegalArgumentException}; semantic constraints are checked by
{@link #validate()}.
+ */
+ public static DynamicAllocationConfig parse(ResourceGroup group) {
+ Map<String, String> properties = group.getProperties();
+ boolean enabled =
+ PropertyUtil.propertyAsBoolean(
+ properties,
+ OptimizerProperties.DYNAMIC_ALLOCATION_ENABLED,
+ OptimizerProperties.DYNAMIC_ALLOCATION_ENABLED_DEFAULT);
+ Integer minParallelism = parseMinParallelism(group);
+
+ Integer maxParallelism =
+ PropertyUtil.propertyAsNullableInt(
+ properties,
OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM);
+
+ return new DynamicAllocationConfig(
+ group.getName(),
+ group.getContainer(),
+ enabled,
+ minParallelism,
+ maxParallelism,
+ parseDuration(
+ properties,
+ OptimizerProperties.DYNAMIC_ALLOCATION_SCHEDULER_BACKLOG_TIMEOUT,
+
OptimizerProperties.DYNAMIC_ALLOCATION_SCHEDULER_BACKLOG_TIMEOUT_DEFAULT),
+ parseDuration(
+ properties,
+ OptimizerProperties.DYNAMIC_ALLOCATION_SUSTAINED_BACKLOG_TIMEOUT,
+
OptimizerProperties.DYNAMIC_ALLOCATION_SUSTAINED_BACKLOG_TIMEOUT_DEFAULT),
+ parseDuration(
+ properties,
+ OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_IDLE_TIMEOUT,
+
OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_IDLE_TIMEOUT_DEFAULT),
+ parseDuration(
+ properties,
+ OptimizerProperties.DYNAMIC_ALLOCATION_SCALE_DOWN_COOLDOWN,
+
OptimizerProperties.DYNAMIC_ALLOCATION_SCALE_DOWN_COOLDOWN_DEFAULT),
+ parseDuration(
+ properties,
+ OptimizerProperties.DYNAMIC_ALLOCATION_DRAIN_TIMEOUT,
+ OptimizerProperties.DYNAMIC_ALLOCATION_DRAIN_TIMEOUT_DEFAULT));
+ }
+
+ /**
+ * Resolve the effective min-parallelism of a group, honoring the deprecated
flat {@code
+ * min-parallelism} as a fallback. Resolution order: {@link
+ * OptimizerProperties#DYNAMIC_ALLOCATION_MIN_PARALLELISM} → {@link
+ * OptimizerProperties#OPTIMIZER_GROUP_MIN_PARALLELISM} → {@code 0}.
Lenient: an unparsable value
+ * falls back to {@code 0} rather than throwing, preserving legacy behavior.
This is on the keeper
+ * hot path and therefore stays silent; deprecation is reported by {@link
+ * #warnDeprecatedMinParallelism(ResourceGroup)} at config-entry points
instead.
+ */
+ public static int resolveMinParallelism(ResourceGroup group) {
+ Map<String, String> properties = group.getProperties();
+ String namespaced =
properties.get(OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM);
+ if (namespaced != null) {
+ return parseIntOrDefault(group.getName(), namespaced);
+ }
+ String legacy =
properties.get(OptimizerProperties.OPTIMIZER_GROUP_MIN_PARALLELISM);
+ if (legacy != null) {
+ return parseIntOrDefault(group.getName(), legacy);
+ }
+ return OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM_DEFAULT;
+ }
+
+ /**
+ * Strictly parse the configured min-parallelism, or {@code null} when
unset. Follows the
+ * resolution order of {@link #resolveMinParallelism(ResourceGroup)} ({@link
+ * OptimizerProperties#DYNAMIC_ALLOCATION_MIN_PARALLELISM} → {@link
+ * OptimizerProperties#OPTIMIZER_GROUP_MIN_PARALLELISM}). Mirrors {@code
max-parallelism}: a
+ * malformed value throws at parse time (honoring {@link #parse}'s contract)
rather than being
+ * silently degraded to {@code 0} by the lenient {@link
#resolveMinParallelism(ResourceGroup)} on
+ * the keeper hot path. Shares its trim semantics so the value {@link
#validate()} accepts is the
+ * one the keeper resolves; the error cites the key the value actually came
from.
+ */
+ private static Integer parseMinParallelism(ResourceGroup group) {
+ Map<String, String> properties = group.getProperties();
+ String key = OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM;
+ String raw = properties.get(key);
+ if (raw == null) {
+ key = OptimizerProperties.OPTIMIZER_GROUP_MIN_PARALLELISM;
+ raw = properties.get(key);
+ }
+ if (raw == null) {
+ return null;
+ }
+ try {
+ return Integer.parseInt(raw.trim());
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Resource group:%s '%s'(%s) is not a valid integer.",
group.getName(), key, raw));
+ }
+ }
+
+ /**
+ * Log a one-off deprecation warning when a group still relies on the flat
{@code
+ * min-parallelism}. Intended for config-entry points (startup load, REST
create/update), not the
+ * keeper hot path.
+ */
+ public static void warnDeprecatedMinParallelism(ResourceGroup group) {
+ Map<String, String> properties = group.getProperties();
+ boolean hasNamespaced =
+
properties.containsKey(OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM);
+ boolean hasLegacy =
properties.containsKey(OptimizerProperties.OPTIMIZER_GROUP_MIN_PARALLELISM);
+ if (hasNamespaced && hasLegacy) {
+ LOG.warn(
+ "Resource group:{} sets both '{}' and the deprecated '{}'; the
namespaced value wins.",
+ group.getName(),
+ OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM,
+ OptimizerProperties.OPTIMIZER_GROUP_MIN_PARALLELISM);
+ } else if (hasLegacy) {
+ LOG.warn(
+ "Resource group:{} uses the deprecated '{}'; please migrate to
'{}'.",
+ group.getName(),
+ OptimizerProperties.OPTIMIZER_GROUP_MIN_PARALLELISM,
+ OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM);
+ }
+ }
+
+ /**
+ * Whether dynamic allocation is effectively enabled for the group: opted in
and the configuration
+ * is valid. An invalid configuration counts as disabled, mirroring the
startup fail-safe
+ * behavior.
+ */
+ public static boolean isEffectivelyEnabled(ResourceGroup group) {
+ try {
+ DynamicAllocationConfig config = parse(group);
+ config.validate();
+ return config.isEnabled();
+ } catch (IllegalArgumentException e) {
+ return false;
+ }
+ }
+
+ /**
+ * The min-parallelism property key that {@link
#resolveMinParallelism(ResourceGroup)} actually
+ * reads for this group. Writers updating the effective value (e.g. the
keeper's auto-reset) must
+ * target this key; writing the deprecated flat key while the namespaced one
is present would be
+ * shadowed.
+ */
+ public static String effectiveMinParallelismKey(ResourceGroup group) {
+ return
group.getProperties().containsKey(OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM)
+ ? OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM
+ : OptimizerProperties.OPTIMIZER_GROUP_MIN_PARALLELISM;
+ }
+
+ /** Enforce the AIP-5 constraints. No-op when DRA is disabled. */
+ public void validate() {
+ if (!enabled) {
+ return;
+ }
+ if (Constants.EXTERNAL_RESOURCE_CONTAINER.equals(container)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Resource group:%s cannot enable dynamic allocation on an
externally-registered "
+ + "optimizer (container '%s'); AMS cannot scale optimizers
it did not launch.",
+ groupName, container));
+ }
+ if (maxParallelism == null) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Resource group:%s enables dynamic allocation but '%s' is not
set; it is required.",
+ groupName,
OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM));
+ }
+ // min-parallelism: a malformed value already threw in parse(); an
opted-in group must also
+ // reject a negative floor. resolveMinParallelism() stays lenient on the
keeper hot path.
+ int minParallelism =
+ this.minParallelism == null
+ ? OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM_DEFAULT
+ : this.minParallelism;
+ if (minParallelism < 0) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Resource group:%s '%s'(%d) must not be negative.",
+ groupName,
OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM, minParallelism));
+ }
+ if (maxParallelism < minParallelism) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Resource group:%s '%s'(%d) must be >= '%s'(%d).",
+ groupName,
+ OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM,
+ maxParallelism,
+ OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM,
+ minParallelism));
+ }
+ if (maxParallelism >
OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM_LIMIT) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Resource group:%s '%s'(%d) must not exceed the hard limit %d.",
+ groupName,
+ OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM,
+ maxParallelism,
+ OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM_LIMIT));
+ }
+ Duration idleMin =
+ ConfigHelpers.TimeUtils.parseDuration(
+ OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_IDLE_TIMEOUT_MIN);
+ if (executorIdleTimeout.compareTo(idleMin) < 0) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Resource group:%s '%s'(%s) must be >= %s.",
+ groupName,
+ OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_IDLE_TIMEOUT,
+ executorIdleTimeout,
+
OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_IDLE_TIMEOUT_MIN));
+ }
+ // TODO AIP-5 Phase 2: these polling-oriented timeouts only enforce a
positive lower bound; a
+ // value like "1ns" passes here but would busy-spin the scale loop. Add
realistic minimums once
+ // the scaling loop that consumes them lands.
+ requirePositive(
+ OptimizerProperties.DYNAMIC_ALLOCATION_SCHEDULER_BACKLOG_TIMEOUT,
schedulerBacklogTimeout);
+ requirePositive(
+ OptimizerProperties.DYNAMIC_ALLOCATION_SUSTAINED_BACKLOG_TIMEOUT,
sustainedBacklogTimeout);
+
requirePositive(OptimizerProperties.DYNAMIC_ALLOCATION_SCALE_DOWN_COOLDOWN,
scaleDownCooldown);
+ requirePositive(OptimizerProperties.DYNAMIC_ALLOCATION_DRAIN_TIMEOUT,
drainTimeout);
+ }
+
+ private void requirePositive(String property, Duration value) {
+ if (value.isZero() || value.isNegative()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Resource group:%s '%s'(%s) must be a positive duration.",
+ groupName, property, value));
+ }
+ }
+
+ private static Duration parseDuration(
+ Map<String, String> properties, String property, String defaultValue) {
+ String value = properties.getOrDefault(property, defaultValue);
+ return ConfigHelpers.TimeUtils.parseDuration(value);
+ }
+
+ private static int parseIntOrDefault(String groupName, String value) {
+ try {
+ return Integer.parseInt(value.trim());
+ } catch (NumberFormatException e) {
+ LOG.warn(
+ "Resource group:{} has an illegal min-parallelism value '{}', using
default {}.",
+ groupName,
+ value,
+ OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM_DEFAULT);
+ return OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM_DEFAULT;
+ }
+ }
+
+ public boolean isEnabled() {
+ return enabled;
+ }
+
+ public int getMinParallelism() {
+ return minParallelism == null
+ ? OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM_DEFAULT
+ : minParallelism;
+ }
+
+ /**
+ * The configured max-parallelism. Precondition: only call on a config that
has passed {@link
+ * #validate()} with DRA enabled — {@code max-parallelism} is nullable and
only mandatory when
+ * enabled, so calling this on a disabled or unvalidated config may throw
{@link
+ * NullPointerException} on unboxing.
+ */
+ public int getMaxParallelism() {
+ return maxParallelism;
+ }
+
+ public Duration getSchedulerBacklogTimeout() {
+ return schedulerBacklogTimeout;
+ }
+
+ public Duration getSustainedBacklogTimeout() {
+ return sustainedBacklogTimeout;
+ }
+
+ public Duration getExecutorIdleTimeout() {
+ return executorIdleTimeout;
+ }
+
+ public Duration getScaleDownCooldown() {
+ return scaleDownCooldown;
+ }
+
+ public Duration getDrainTimeout() {
+ return drainTimeout;
+ }
+}
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerGroupController.java
b/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerGroupController.java
index 6e40077c7..a4dfa4290 100644
---
a/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerGroupController.java
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerGroupController.java
@@ -19,14 +19,19 @@
package org.apache.amoro.server;
import static org.junit.Assert.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import io.javalin.http.Context;
+import org.apache.amoro.OptimizerProperties;
+import org.apache.amoro.resource.ResourceGroup;
import org.apache.amoro.server.dashboard.controller.OptimizerGroupController;
import org.apache.amoro.server.resource.OptimizerManager;
import org.apache.amoro.server.table.TableManager;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.MethodSource;
@@ -77,4 +82,43 @@ public class TestOptimizerGroupController {
when(ctx.bodyAsClass(Map.class)).thenReturn(requestBody);
assertThrows(BadRequestException.class, () ->
controller.createResourceGroup(ctx));
}
+
+ private Map<String, Object> groupRequest(String name, Map<String, String>
properties) {
+ Map<String, Object> requestBody = new HashMap<>();
+ requestBody.put("name", name);
+ requestBody.put("container", "flink");
+ requestBody.put("properties", properties);
+ return requestBody;
+ }
+
+ @Test
+ void createWithInvalidDynamicAllocationIsRejected() {
+ Map<String, String> properties = new HashMap<>();
+ // enabled without the required max-parallelism
+ properties.put(OptimizerProperties.DYNAMIC_ALLOCATION_ENABLED, "true");
+
+ when(ctx.bodyAsClass(Map.class)).thenReturn(groupRequest("group1",
properties));
+ assertThrows(BadRequestException.class, () ->
controller.createResourceGroup(ctx));
+ }
+
+ @Test
+ void updateWithInvalidDynamicAllocationIsRejected() {
+ Map<String, String> properties = new HashMap<>();
+ properties.put(OptimizerProperties.DYNAMIC_ALLOCATION_ENABLED, "true");
+ properties.put(OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM,
"2048");
+
+ when(ctx.bodyAsClass(Map.class)).thenReturn(groupRequest("group1",
properties));
+ assertThrows(BadRequestException.class, () ->
controller.updateResourceGroup(ctx));
+ }
+
+ @Test
+ void createWithValidDynamicAllocationSucceeds() {
+ Map<String, String> properties = new HashMap<>();
+ properties.put(OptimizerProperties.DYNAMIC_ALLOCATION_ENABLED, "true");
+ properties.put(OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM,
"16");
+
+ when(ctx.bodyAsClass(Map.class)).thenReturn(groupRequest("group1",
properties));
+ controller.createResourceGroup(ctx);
+ verify(optimizerManager).createResourceGroup(any(ResourceGroup.class));
+ }
}
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerGroupKeeper.java
b/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerGroupKeeper.java
index 889fb02cc..20d4504cc 100644
---
a/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerGroupKeeper.java
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerGroupKeeper.java
@@ -313,6 +313,67 @@ public class TestOptimizerGroupKeeper extends
AMSTableTestBase {
+ ":min-parallelism should be reset to optimizer's current total
cores (1) when no more resources available");
}
+ /**
+ * Test scenario 5: min-parallelism auto-reset is skipped when dynamic
allocation is enabled.
+ *
+ * <p>When a group opts into dynamic allocation, DRA owns scale decisions
for the group; the
+ * keeper must not erode its min-parallelism floor even after exhausting
creation attempts.
+ */
+ @Test
+ public void testAutoResetSkippedWhenDynamicAllocationEnabled() throws
InterruptedException {
+ resourceAvailable.set(false);
+ scaleOutCallCount.set(0);
+ ResourceGroup resourceGroup = buildTestResourceGroup(TEST_GROUP_NAME +
"-5", 2);
+
resourceGroup.getProperties().put(OptimizerProperties.DYNAMIC_ALLOCATION_ENABLED,
"true");
+
resourceGroup.getProperties().put(OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM,
"8");
+
+ optimizerManager().createResourceGroup(resourceGroup);
+ optimizingService().createResourceGroup(resourceGroup);
+
+ Thread.sleep(200);
+
+ ResourceGroup updatedGroup =
optimizerManager().getResourceGroup(resourceGroup.getName());
+ Assertions.assertEquals(
+ "2",
+
updatedGroup.getProperties().get(OptimizerProperties.OPTIMIZER_GROUP_MIN_PARALLELISM),
+ resourceGroup.getName()
+ + ":min-parallelism must not be auto-reset when dynamic allocation
is enabled");
+ }
+
+ /**
+ * Test scenario 6: auto-reset writes the namespaced min-parallelism key
when the group uses it.
+ *
+ * <p>A group configured with dynamic-allocation.min-parallelism (DRA not
enabled) must have the
+ * auto-reset applied to that key; writing the deprecated flat key would be
shadowed by the
+ * namespaced one, turning the reset into an endless no-op loop.
+ */
+ @Test
+ public void testAutoResetWritesNamespacedKeyWhenUsed() throws
InterruptedException {
+ resourceAvailable.set(false);
+ scaleOutCallCount.set(0);
+ String groupName = TEST_GROUP_NAME + "-6";
+ this.currentGroupName = groupName;
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM,
"2");
+ properties.put("memory", "1024");
+ ResourceGroup resourceGroup =
+ new ResourceGroup.Builder(groupName,
MOCK_CONTAINER_NAME).addProperties(properties).build();
+
+ optimizerManager().createResourceGroup(resourceGroup);
+ optimizingService().createResourceGroup(resourceGroup);
+
+ Thread.sleep(200);
+
+ ResourceGroup updatedGroup =
optimizerManager().getResourceGroup(resourceGroup.getName());
+ Assertions.assertEquals(
+ "0",
+
updatedGroup.getProperties().get(OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM),
+ groupName + ":auto-reset must write the namespaced min-parallelism key
the group uses");
+ Assertions.assertNull(
+
updatedGroup.getProperties().get(OptimizerProperties.OPTIMIZER_GROUP_MIN_PARALLELISM),
+ groupName + ":auto-reset must not introduce the deprecated flat
min-parallelism key");
+ }
+
private static OptimizerRegisterInfo buildRegisterInfo(String groupName, int
threadCount) {
OptimizerRegisterInfo registerInfo = new OptimizerRegisterInfo();
Map<String, String> registerProperties = Maps.newHashMap();
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
new file mode 100644
index 000000000..837584285
--- /dev/null
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestDynamicAllocationConfig.java
@@ -0,0 +1,245 @@
+/*
+ * 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 static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+
+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.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+
+public class TestDynamicAllocationConfig {
+
+ private static final String CONTAINER = "flink";
+
+ private ResourceGroup group(Map<String, String> properties) {
+ return new ResourceGroup.Builder("group1",
CONTAINER).addProperties(properties).build();
+ }
+
+ private Map<String, String> enabledProps() {
+ Map<String, String> props = new HashMap<>();
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_ENABLED, "true");
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM, "16");
+ return props;
+ }
+
+ private void parseAndValidate(ResourceGroup group) {
+ DynamicAllocationConfig.parse(group).validate();
+ }
+
+ @Test
+ void enabledWithoutMaxParallelismIsRejected() {
+ Map<String, String> props = new HashMap<>();
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_ENABLED, "true");
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
parseAndValidate(group(props)));
+ }
+
+ @Test
+ void maxParallelismBelowMinIsRejected() {
+ Map<String, String> props = enabledProps();
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM, "32");
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM, "16");
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
parseAndValidate(group(props)));
+ }
+
+ @Test
+ void maxParallelismAboveHardLimitIsRejected() {
+ Map<String, String> props = enabledProps();
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM, "2048");
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
parseAndValidate(group(props)));
+ }
+
+ @Test
+ void enabledWithUnparsableMinParallelismIsRejected() {
+ // resolveMinParallelism() is lenient (legacy/keeper path), but an
opted-in group must not
+ // silently degrade an unparsable min-parallelism to 0.
+ Map<String, String> props = enabledProps();
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM, "abc");
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
parseAndValidate(group(props)));
+ }
+
+ @Test
+ void enabledWithNegativeMinParallelismIsRejected() {
+ Map<String, String> props = enabledProps();
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM, "-3");
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
parseAndValidate(group(props)));
+ }
+
+ @Test
+ void malformedMinParallelismIsRejectedAtParseEvenWhenDisabled() {
+ // Mirrors max-parallelism: a malformed numeric is rejected at parse()
regardless of enabled,
+ // honoring parse()'s documented contract rather than being silently
degraded to 0.
+ Map<String, String> props = new HashMap<>();
+ props.put(OptimizerProperties.OPTIMIZER_GROUP_MIN_PARALLELISM, "abc");
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
DynamicAllocationConfig.parse(group(props)));
+ }
+
+ @Test
+ void malformedMinParallelismErrorCitesTheKeyActuallyUsed() {
+ // Value comes only from the legacy key; the error must name that key, not
the namespaced one
+ // the user never set.
+ Map<String, String> props = new HashMap<>();
+ props.put(OptimizerProperties.OPTIMIZER_GROUP_MIN_PARALLELISM, "abc");
+ IllegalArgumentException e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
DynamicAllocationConfig.parse(group(props)));
+ Assertions.assertTrue(
+ e.getMessage().contains("'" +
OptimizerProperties.OPTIMIZER_GROUP_MIN_PARALLELISM + "'"),
+ "error should cite the legacy key actually used: " + e.getMessage());
+ }
+
+ @Test
+ void whitespacePaddedMinParallelismIsResolvedConsistently() {
+ // validate() trims, so it accepts " 5 " as 5; the keeper's
resolveMinParallelism() must agree.
+ // Otherwise a config validate() accepts silently degrades to a floor of 0
at runtime.
+ Map<String, String> props = enabledProps();
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM, " 5 ");
+ ResourceGroup g = group(props);
+
+ DynamicAllocationConfig config = DynamicAllocationConfig.parse(g);
+ assertDoesNotThrow(config::validate);
+ Assertions.assertEquals(5, config.getMinParallelism());
+ Assertions.assertEquals(5,
DynamicAllocationConfig.resolveMinParallelism(g));
+ }
+
+ @Test
+ void executorIdleTimeoutBelowMinimumIsRejected() {
+ Map<String, String> props = enabledProps();
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_IDLE_TIMEOUT,
"10s");
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
parseAndValidate(group(props)));
+ }
+
+ @Test
+ void unparsableDurationIsRejected() {
+ Map<String, String> props = enabledProps();
+
props.put(OptimizerProperties.DYNAMIC_ALLOCATION_SCHEDULER_BACKLOG_TIMEOUT,
"not-a-duration");
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
parseAndValidate(group(props)));
+ }
+
+ @Test
+ void zeroDurationIsRejected() {
+ Map<String, String> props = enabledProps();
+
props.put(OptimizerProperties.DYNAMIC_ALLOCATION_SCHEDULER_BACKLOG_TIMEOUT,
"0s");
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
parseAndValidate(group(props)));
+ }
+
+ @Test
+ void externalContainerWithEnabledIsRejected() {
+ ResourceGroup external =
+ new
ResourceGroup.Builder("group1").addProperties(enabledProps()).build();
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
DynamicAllocationConfig.parse(external).validate());
+ }
+
+ @Test
+ void validConfigIsParsedCorrectly() {
+ Map<String, String> props = enabledProps();
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM, "4");
+
props.put(OptimizerProperties.DYNAMIC_ALLOCATION_SCHEDULER_BACKLOG_TIMEOUT,
"2min");
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_IDLE_TIMEOUT,
"10min");
+
+ DynamicAllocationConfig config =
DynamicAllocationConfig.parse(group(props));
+ assertDoesNotThrow(config::validate);
+
+ Assertions.assertTrue(config.isEnabled());
+ Assertions.assertEquals(4, config.getMinParallelism());
+ Assertions.assertEquals(16, config.getMaxParallelism());
+ Assertions.assertEquals(Duration.ofMinutes(2),
config.getSchedulerBacklogTimeout());
+ Assertions.assertEquals(Duration.ofMinutes(10),
config.getExecutorIdleTimeout());
+ Assertions.assertEquals(Duration.ofSeconds(30),
config.getSustainedBacklogTimeout());
+ Assertions.assertEquals(Duration.ofMinutes(1),
config.getScaleDownCooldown());
+ Assertions.assertEquals(Duration.ofMinutes(15), config.getDrainTimeout());
+ }
+
+ @Test
+ void minParallelismResolutionPrefersNamespacedValue() {
+ Map<String, String> props = enabledProps();
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM, "8");
+ props.put(OptimizerProperties.OPTIMIZER_GROUP_MIN_PARALLELISM, "2");
+ Assertions.assertEquals(8,
DynamicAllocationConfig.resolveMinParallelism(group(props)));
+ }
+
+ @Test
+ void minParallelismResolutionFallsBackToLegacyValue() {
+ Map<String, String> props = new HashMap<>();
+ props.put(OptimizerProperties.OPTIMIZER_GROUP_MIN_PARALLELISM, "5");
+ Assertions.assertEquals(5,
DynamicAllocationConfig.resolveMinParallelism(group(props)));
+ }
+
+ @Test
+ void minParallelismResolutionDefaultsToZero() {
+ Assertions.assertEquals(
+ 0, DynamicAllocationConfig.resolveMinParallelism(group(new
HashMap<>())));
+ }
+
+ @Test
+ void disabledConfigSkipsValidation() {
+ Map<String, String> props = new HashMap<>();
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_ENABLED, "false");
+ // max-parallelism absent and an external container would fail if
validated, but disabled.
+ ResourceGroup external = new
ResourceGroup.Builder("group1").addProperties(props).build();
+ assertDoesNotThrow(() ->
DynamicAllocationConfig.parse(external).validate());
+ }
+
+ @Test
+ void effectivelyEnabledWhenValidConfig() {
+
Assertions.assertTrue(DynamicAllocationConfig.isEffectivelyEnabled(group(enabledProps())));
+ }
+
+ @Test
+ void notEffectivelyEnabledWhenDisabled() {
+
Assertions.assertFalse(DynamicAllocationConfig.isEffectivelyEnabled(group(new
HashMap<>())));
+ }
+
+ @Test
+ void notEffectivelyEnabledWhenConfigInvalid() {
+ Map<String, String> props = new HashMap<>();
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_ENABLED, "true");
+ // enabled without max-parallelism is invalid, so DRA falls back to
disabled.
+
Assertions.assertFalse(DynamicAllocationConfig.isEffectivelyEnabled(group(props)));
+ }
+
+ @Test
+ void effectiveMinParallelismKeyPrefersNamespacedWhenPresent() {
+ Map<String, String> props = new HashMap<>();
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM, "2");
+ props.put(OptimizerProperties.OPTIMIZER_GROUP_MIN_PARALLELISM, "2");
+ Assertions.assertEquals(
+ OptimizerProperties.DYNAMIC_ALLOCATION_MIN_PARALLELISM,
+ DynamicAllocationConfig.effectiveMinParallelismKey(group(props)));
+ }
+
+ @Test
+ void effectiveMinParallelismKeyFallsBackToLegacy() {
+ Map<String, String> props = new HashMap<>();
+ props.put(OptimizerProperties.OPTIMIZER_GROUP_MIN_PARALLELISM, "2");
+ Assertions.assertEquals(
+ OptimizerProperties.OPTIMIZER_GROUP_MIN_PARALLELISM,
+ DynamicAllocationConfig.effectiveMinParallelismKey(group(props)));
+ Assertions.assertEquals(
+ OptimizerProperties.OPTIMIZER_GROUP_MIN_PARALLELISM,
+ DynamicAllocationConfig.effectiveMinParallelismKey(group(new
HashMap<>())));
+ }
+}
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 3c5e7f900..c69a8aa94 100644
--- a/amoro-common/src/main/java/org/apache/amoro/OptimizerProperties.java
+++ b/amoro-common/src/main/java/org/apache/amoro/OptimizerProperties.java
@@ -32,7 +32,13 @@ public class OptimizerProperties {
public static final String OPTIMIZER_EXECUTION_PARALLEL =
"execution-parallel";
public static final String OPTIMIZER_MEMORY_SIZE = "memory-size";
public static final String OPTIMIZER_GROUP_NAME = "group-name";
- public static final String OPTIMIZER_GROUP_MIN_PARALLELISM =
"min-parallelism";
+
+ /**
+ * @deprecated since 0.9.0, use {@link #DYNAMIC_ALLOCATION_MIN_PARALLELISM}
instead. Still honored
+ * as a fallback when the namespaced property is absent.
+ */
+ @Deprecated public static final String OPTIMIZER_GROUP_MIN_PARALLELISM =
"min-parallelism";
+
public static final String OPTIMIZER_HEART_BEAT_INTERVAL =
"heart-beat-interval";
public static final String OPTIMIZER_EXTEND_DISK_STORAGE =
"extend-disk-storage";
public static final boolean OPTIMIZER_EXTEND_DISK_STORAGE_DEFAULT = false;
@@ -51,4 +57,54 @@ public class OptimizerProperties {
public static final String OPTIMIZER_MASTER_SLAVE_MODE_ENABLED =
"master-slave-mode-enabled";
public static final String OPTIMIZER_SHUTDOWN_TIMEOUT_MS =
"shutdown-timeout-ms";
public static final long OPTIMIZER_SHUTDOWN_TIMEOUT_MS_DEFAULT = 600_000L;
// 10 min
+
+ // Dynamic resource allocation (DRA) properties (AIP-5), configured at the
resource group level.
+ // Semantics and validation rules are documented in DynamicAllocationConfig
(amoro-ams).
+
+ /** @since 0.9.0 */
+ public static final String DYNAMIC_ALLOCATION_ENABLED =
"dynamic-allocation.enabled";
+
+ public static final boolean DYNAMIC_ALLOCATION_ENABLED_DEFAULT = false;
+
+ /** @since 0.9.0 */
+ public static final String DYNAMIC_ALLOCATION_MIN_PARALLELISM =
+ "dynamic-allocation.min-parallelism";
+
+ public static final int DYNAMIC_ALLOCATION_MIN_PARALLELISM_DEFAULT = 0;
+
+ /** @since 0.9.0 */
+ public static final String DYNAMIC_ALLOCATION_MAX_PARALLELISM =
+ "dynamic-allocation.max-parallelism";
+
+ public static final int DYNAMIC_ALLOCATION_MAX_PARALLELISM_LIMIT = 1024;
+
+ /** @since 0.9.0 */
+ public static final String DYNAMIC_ALLOCATION_SCHEDULER_BACKLOG_TIMEOUT =
+ "dynamic-allocation.scheduler-backlog-timeout";
+
+ public static final String
DYNAMIC_ALLOCATION_SCHEDULER_BACKLOG_TIMEOUT_DEFAULT = "1min";
+
+ /** @since 0.9.0 */
+ public static final String DYNAMIC_ALLOCATION_SUSTAINED_BACKLOG_TIMEOUT =
+ "dynamic-allocation.sustained-backlog-timeout";
+
+ public static final String
DYNAMIC_ALLOCATION_SUSTAINED_BACKLOG_TIMEOUT_DEFAULT = "30s";
+
+ /** @since 0.9.0 */
+ public static final String DYNAMIC_ALLOCATION_EXECUTOR_IDLE_TIMEOUT =
+ "dynamic-allocation.executor-idle-timeout";
+
+ public static final String DYNAMIC_ALLOCATION_EXECUTOR_IDLE_TIMEOUT_DEFAULT
= "5min";
+ public static final String DYNAMIC_ALLOCATION_EXECUTOR_IDLE_TIMEOUT_MIN =
"30s";
+
+ /** @since 0.9.0 */
+ public static final String DYNAMIC_ALLOCATION_SCALE_DOWN_COOLDOWN =
+ "dynamic-allocation.scale-down-cooldown";
+
+ public static final String DYNAMIC_ALLOCATION_SCALE_DOWN_COOLDOWN_DEFAULT =
"1min";
+
+ /** @since 0.9.0 */
+ public static final String DYNAMIC_ALLOCATION_DRAIN_TIMEOUT =
"dynamic-allocation.drain-timeout";
+
+ public static final String DYNAMIC_ALLOCATION_DRAIN_TIMEOUT_DEFAULT =
"15min";
}
