From 84234e45972b2ab43e9e88e50bb8b0d6787ef8ea Mon Sep 17 00:00:00 2001 From: Jiwon Park Date: Sat, 11 Jul 2026 21:16:18 +0900 Subject: [PATCH 01/10] [AMORO-4271] AIP-5 Phase 2: add dynamic-allocation.executor-parallelism config The scaling unit of dynamic allocation is one homogeneous K-thread optimizer instance (the Spark executor model). K is configured by the new dynamic-allocation.executor-parallelism property (default 1). validate() rejects K < 1 and K > max-parallelism: a single K-thread instance already exceeding the cap could never be created, which would leave an enabled group as a silent no-op. Signed-off-by: Jiwon Park --- .../dra/DynamicAllocationConfig.java | 34 +++++++++++++++ .../dra/TestDynamicAllocationConfig.java | 42 +++++++++++++++++++ .../org/apache/amoro/OptimizerProperties.java | 6 +++ 3 files changed, 82 insertions(+) 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..b9ec526465 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,26 @@ 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)); + } Duration idleMin = ConfigHelpers.TimeUtils.parseDuration( OptimizerProperties.DYNAMIC_ALLOCATION_EXECUTOR_IDLE_TIMEOUT_MIN); @@ -340,6 +370,10 @@ public int getMaxParallelism() { return maxParallelism; } + public int getExecutorParallelism() { + return executorParallelism; + } + public Duration getSchedulerBacklogTimeout() { return schedulerBacklogTimeout; } 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..db7374a66c 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,48 @@ 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 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-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"; From e70c794ec776069a958228884c2a91be86709aa3 Mon Sep 17 00:00:00 2001 From: Jiwon Park Date: Sat, 11 Jul 2026 21:21:15 +0900 Subject: [PATCH 02/10] [AMORO-4271] AIP-5 Phase 2: pure demand accounting (serviceable planned, thread occupancy) serviceablePlannedCount implements quota-mode-aware demand counting: a proportional quota (targetQuota <= 1) scales with availableCore, so the whole backlog is serviceable; an absolute quota (> 1) is a fixed limit that scaling cannot raise, so only free slots count. occupiesThread counts SCHEDULED as occupying: a thread is busy from assignment (pollTask), not from ack; counting only ACKED would overestimate headroom during the poll-to-ack window. Signed-off-by: Jiwon Park --- .../dra/DynamicAllocationState.java | 75 ++++++++++ .../dra/TestDynamicAllocationState.java | 132 ++++++++++++++++++ 2 files changed, 207 insertions(+) create mode 100644 amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java create mode 100644 amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestDynamicAllocationState.java 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..7ef98d5aaa --- /dev/null +++ b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java @@ -0,0 +1,75 @@ +/* + * 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; + +/** + * Demand accounting for dynamic allocation (AIP-5). Pure functions over plain values so the scaling + * decision logic is testable without a live optimizing queue. + */ +public final class DynamicAllocationState { + + private DynamicAllocationState() {} + + /** 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; + } +} 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..b1ed79d80f --- /dev/null +++ b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestDynamicAllocationState.java @@ -0,0 +1,132 @@ +/* + * 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 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())); + } + + // --- 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)); + } +} From 6290345e42dfc13123a0a0ad7cfe9399f040ecca Mon Sep 17 00:00:00 2001 From: Jiwon Park Date: Sat, 11 Jul 2026 21:27:23 +0900 Subject: [PATCH 03/10] [AMORO-4271] AIP-5 Phase 2: computeScaleUp scale-out decision Per-group decision state (backlog timer, cadence, exponential ramp) with injected time, so every scenario is deterministic. Ordered checks: - min-parallelism floor: enforced immediately, no timing gate. - Immediate demand (busy + serviceable > effective): exponential ramp (1, 2, 4, 8) clamped to the actual need; the ramp resets when the clamp binds (Spark addExecutors semantics) or when demand clears. - Future demand (pending tables while all threads are busy, including the zero-optimizer cold start where nothing polls and planning never runs): a single probe instance. Pending tables are not quantified demand before planning, so no exponential growth on this signal. Demand must persist for scheduler-backlog-timeout before the first scale-out; later rounds are spaced by sustained-backlog-timeout, so a trickle drained between rounds never accumulates toward a scale-out. The max-parallelism cap always wins. Signed-off-by: Jiwon Park --- .../dra/DynamicAllocationState.java | 91 +++++++- .../optimizing/dra/TestComputeScaleUp.java | 213 ++++++++++++++++++ 2 files changed, 301 insertions(+), 3 deletions(-) create mode 100644 amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestComputeScaleUp.java diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java index 7ef98d5aaa..a218e3e0d0 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java @@ -23,12 +23,97 @@ import java.util.Collection; /** - * Demand accounting for dynamic allocation (AIP-5). Pure functions over plain values so the scaling - * decision logic is testable without a live optimizing queue. + * 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 { - private 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) { + 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; + } /** Per-table demand snapshot consumed by {@link #serviceablePlannedCount(Collection)}. */ public static class TableDemand { 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..08ecd543d7 --- /dev/null +++ b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestComputeScaleUp.java @@ -0,0 +1,213 @@ +/* + * 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. + 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 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)); + } +} From 576dd7edf32aa80baad1c5d517f9210dd58048ad Mon Sep 17 00:00:00 2001 From: Jiwon Park Date: Sat, 11 Jul 2026 21:31:09 +0900 Subject: [PATCH 04/10] [AMORO-4271] AIP-5 Phase 2: pending registration accounting with boot deadline Registration is optimizer-driven (the pod self-registers after boot), so scale-up counts requested-but-unregistered capacity to avoid duplicate scale-outs during the boot window. Entries carry their own boot deadline: a request that never registers (image pull failure, exhausted ResourceQuota, crash loop) is pruned instead of freezing scale-up below real demand forever. Heartbeat expiry cannot cover this window because it only starts after registration. Signed-off-by: Jiwon Park --- .../optimizing/dra/PendingRegistrations.java | 71 +++++++++++++++ .../dra/TestPendingRegistrations.java | 87 +++++++++++++++++++ 2 files changed, 158 insertions(+) create mode 100644 amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/PendingRegistrations.java create mode 100644 amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestPendingRegistrations.java 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/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)); + } +} From 948ef760c113f4589b8e265e2ee65179173af55c Mon Sep 17 00:00:00 2001 From: Jiwon Park Date: Sat, 11 Jul 2026 21:45:50 +0900 Subject: [PATCH 05/10] [AMORO-4271] AIP-5 Phase 2: dedicated scale keeper for dynamic-allocation groups DRA-enabled groups are taken over from the legacy floor keeper, whose min-parallelism-check cadence (minutes, multiplied by consecutive attempts) would render the DRA backlog timeouts (seconds) physically unreachable. OptimizerScaleKeeper reuses the AbstractKeeper infrastructure (HA leader gating, DelayQueue, lifecycle) and evaluates each group at its own sustained-backlog-timeout; the legacy keeper keeps watching DRA groups only to resume floor duty if DRA is disabled later. Scale-outs are executed in executor-parallelism-thread instance units (computeScaleUp), with requested-but-unregistered capacity counted via PendingRegistrations so a booting pod is not re-requested every round; a synchronous request failure is dropped immediately and retried. Groups are watched on create, startup load, and update (covering enabling DRA on an existing group at runtime); deleted or disabled groups drop out of the watch set on their next evaluation. OptimizingQueue.collectDynamicAllocationLoad() snapshots the demand side: busy threads (SCHEDULED and ACKED), quota-mode-aware serviceable PLANNED tasks, and PENDING tables - the only signal observable on a cold group with zero optimizers. Signed-off-by: Jiwon Park --- .../server/DefaultOptimizingService.java | 173 ++++++++++- .../server/optimizing/OptimizingQueue.java | 42 +++ .../dra/DynamicAllocationState.java | 25 ++ .../server/TestOptimizerScaleKeeper.java | 281 ++++++++++++++++++ 4 files changed, 510 insertions(+), 11 deletions(-) create mode 100644 amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerScaleKeeper.java 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..c95d5e68e1 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; @@ -116,6 +118,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 +197,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 +234,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,6 +411,7 @@ public void createResourceGroup(ResourceGroup resourceGroup) { String groupName = resourceGroup.getName(); optimizingQueueByGroup.put(groupName, optimizingQueue); optimizerGroupKeeper.keepInTouch(groupName, 1); + optimizerScaleKeeper.watch(resourceGroup); }); } @@ -416,6 +423,7 @@ public void deleteResourceGroup(String groupName) { public void updateResourceGroup(ResourceGroup resourceGroup) { Optional.ofNullable(optimizingQueueByGroup.get(resourceGroup.getName())) .ifPresent(queue -> queue.updateOptimizerGroup(resourceGroup)); + optimizerScaleKeeper.watch(resourceGroup); } public void dispose() { @@ -426,6 +434,7 @@ public void dispose() { optimizingQueueByGroup.values().forEach(OptimizingQueue::dispose); optimizerKeeper.dispose(); optimizerGroupKeeper.dispose(); + optimizerScaleKeeper.dispose(); tableHandlerChain.dispose(); optimizingQueueByGroup.clear(); optimizingQueueByToken.clear(); @@ -497,6 +506,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 +910,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 +927,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 +963,149 @@ 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; + + private final Map scaleStates = new ConcurrentHashMap<>(); + private final Map pendingRegistrations = + new ConcurrentHashMap<>(); + private final Set watchedGroups = ConcurrentHashMap.newKeySet(); + + 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); + pendingRegistrations.remove(groupName); + } + + @Override + protected void processTask(DraScaleTask task) { + ResourceGroup resourceGroup; + try { + resourceGroup = optimizerManager.getResourceGroup(task.groupName); + } catch (Exception e) { + resourceGroup = null; + } + OptimizingQueue queue = optimizingQueueByGroup.get(task.groupName); + if (resourceGroup == null + || queue == null + || !DynamicAllocationConfig.isEffectivelyEnabled(resourceGroup)) { + // Deleted or disabled: stop watching; an update re-enabling DRA re-watches the group. + unwatch(task.groupName); + 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())); + } + } + + 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 effectiveThreads = getTotalQuota(groupName) + pending.pendingThreads(now); + DynamicAllocationState.GroupLoad load = queue.collectDynamicAllocationLoad(); + 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); + try { + ((AbstractOptimizerContainer) resourceContainer).requestResource(resource); + optimizerManager.createResource(resource); + } catch (Throwable t) { + 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..b74562fb95 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.getTableRuntimeMap().values()) { + 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/dra/DynamicAllocationState.java b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java index a218e3e0d0..dd7ca50696 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java @@ -115,6 +115,31 @@ 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; 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..dca047f178 --- /dev/null +++ b/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerScaleKeeper.java @@ -0,0 +1,281 @@ +/* + * 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")); + } + + /** + * 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"); + } +} From d07b8542e08b1d80a86ebe1f6344436cabe092b0 Mon Sep 17 00:00:00 2001 From: Jiwon Park Date: Sat, 11 Jul 2026 21:50:32 +0900 Subject: [PATCH 06/10] [AMORO-4271] AIP-5 Phase 2: warn when a group is planning-bound instead of scaling Idle threads while tables wait as PENDING and no PLANNED tasks materialize means the bottleneck is the serialized planning (optimizer.max-planning-parallelism), not thread capacity: scaling out would only add more idle threads. The scale keeper surfaces this as an edge-triggered warning naming the config to raise, and does not scale. A cold group (zero threads) stays the future-demand case. Signed-off-by: Jiwon Park --- .../server/DefaultOptimizingService.java | 32 ++++++++++++++++++ .../dra/DynamicAllocationState.java | 15 +++++++++ .../dra/TestDynamicAllocationState.java | 33 +++++++++++++++++++ 3 files changed, 80 insertions(+) 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 c95d5e68e1..2e2b4f4a3b 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 @@ -1002,6 +1002,7 @@ private class OptimizerScaleKeeper extends AbstractKeeper { private final Map pendingRegistrations = new ConcurrentHashMap<>(); private final Set watchedGroups = ConcurrentHashMap.newKeySet(); + private final Set planningBoundGroups = ConcurrentHashMap.newKeySet(); public OptimizerScaleKeeper(String threadName) { super(threadName); @@ -1032,6 +1033,7 @@ private void unwatch(String groupName) { watchedGroups.remove(groupName); scaleStates.remove(groupName); pendingRegistrations.remove(groupName); + planningBoundGroups.remove(groupName); } @Override @@ -1061,6 +1063,35 @@ protected void processTask(DraScaleTask task) { } } + /** + * Warn once on entering 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. Edge-triggered to avoid a warning per evaluation round. + */ + private void warnOnPlanningBoundTransition( + String groupName, int effectiveThreads, DynamicAllocationState.GroupLoad load) { + boolean planningBound = + DynamicAllocationState.isPlanningBound( + effectiveThreads, + load.getBusyThreads(), + load.getServiceablePlanned(), + load.getPendingTables()); + if (planningBound) { + if (planningBoundGroups.add(groupName)) { + 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, + effectiveThreads - load.getBusyThreads(), + load.getPendingTables(), + AmoroManagementConf.OPTIMIZER_MAX_PLANNING_PARALLELISM.key()); + } + } else { + planningBoundGroups.remove(groupName); + } + } + private void scaleIfNeeded( ResourceGroup resourceGroup, OptimizingQueue queue, DynamicAllocationConfig config) { String groupName = resourceGroup.getName(); @@ -1072,6 +1103,7 @@ private void scaleIfNeeded( scaleStates.computeIfAbsent(groupName, name -> new DynamicAllocationState()); int effectiveThreads = getTotalQuota(groupName) + pending.pendingThreads(now); DynamicAllocationState.GroupLoad load = queue.collectDynamicAllocationLoad(); + warnOnPlanningBoundTransition(groupName, effectiveThreads, load); int addInstances = state.computeScaleUp( effectiveThreads, diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java index dd7ca50696..badb21f04d 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java @@ -182,4 +182,19 @@ public static int serviceablePlannedCount(Collection demands) { 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/test/java/org/apache/amoro/server/optimizing/dra/TestDynamicAllocationState.java b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestDynamicAllocationState.java index b1ed79d80f..b5837fa2ab 100644 --- 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 @@ -107,6 +107,39 @@ void emptyDemandsCountZero() { 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 From 399b69c2a7d010b60aaedc8bde42957385b4e2cf Mon Sep 17 00:00:00 2001 From: Jiwon Park Date: Sun, 12 Jul 2026 19:09:07 +0900 Subject: [PATCH 07/10] [AMORO-4271] AIP-5 Phase 2: distinguish the scale keeper's failure modes The keeper's drop-out path treated every irregularity the same way, which conflates situations that need opposite handling: - A transient group-read failure (e.g. a database hiccup) is not a deletion: there is no periodic re-watch, so dropping the group there would silently disable its dynamic allocation until the next config change. Keep the task alive and retry shortly. - After unwatching a disabled group, re-read it once: an update re-enabling DRA concurrently would have had its watch() call swallowed by the still-present watched-set entry, orphaning the group until its next change. - An enabled group whose queue is momentarily absent (a delete/create racing the config watcher) is transient too; unwatch-plus-rewatch there would spin a delay-0 hot loop of DB reads. - A scale-out that fails after requestResource succeeded has started a real pod; erasing its boot-window entry would re-request a duplicate next round. Only a failure before the request is dropped and retried. - Disabling DRA keeps the boot-window accounting (re-enabling within the window must not re-request the same capacity); deleting the group drops everything immediately, so a same-name group created before the next evaluation does not inherit phantom capacity. - The planning-bound warning now counts registered threads only (a booting pod's phantom capacity is not idle threads) and requires the condition to persist across two consecutive evaluations, since a single snapshot can hold transiently while planning is in flight. Also pins in the integration test that registration clears the boot-window accounting: double-counted capacity would suppress demand scaling. Signed-off-by: Jiwon Park --- .../server/DefaultOptimizingService.java | 134 ++++++++++++++---- .../server/TestOptimizerScaleKeeper.java | 5 + 2 files changed, 110 insertions(+), 29 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 2e2b4f4a3b..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 @@ -61,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; @@ -418,6 +419,7 @@ public void createResourceGroup(ResourceGroup resourceGroup) { public void deleteResourceGroup(String groupName) { OptimizingQueue optimizingQueue = optimizingQueueByGroup.remove(groupName); optimizingQueue.dispose(); + optimizerScaleKeeper.onGroupDeleted(groupName); } public void updateResourceGroup(ResourceGroup resourceGroup) { @@ -426,6 +428,11 @@ public void updateResourceGroup(ResourceGroup resourceGroup) { optimizerScaleKeeper.watch(resourceGroup); } + @VisibleForTesting + int pendingScaleThreads(String groupName) { + return optimizerScaleKeeper.pendingThreads(groupName); + } + public void dispose() { planExecutor.shutdown(); // shutdown sync group first, stop syncing group @@ -998,11 +1005,15 @@ private class OptimizerScaleKeeper extends AbstractKeeper { // 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 Set planningBoundGroups = ConcurrentHashMap.newKeySet(); + private final Map planningBoundStreaks = new ConcurrentHashMap<>(); public OptimizerScaleKeeper(String threadName) { super(threadName); @@ -1032,8 +1043,20 @@ public void onOptimizerRegistered(OptimizerInstance optimizer) { 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); - planningBoundGroups.remove(groupName); } @Override @@ -1042,14 +1065,31 @@ protected void processTask(DraScaleTask task) { try { resourceGroup = optimizerManager.getResourceGroup(task.groupName); } catch (Exception e) { - resourceGroup = null; + // 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; } - OptimizingQueue queue = optimizingQueueByGroup.get(task.groupName); - if (resourceGroup == null - || queue == null - || !DynamicAllocationConfig.isEffectivelyEnabled(resourceGroup)) { + 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); @@ -1063,32 +1103,54 @@ protected void processTask(DraScaleTask task) { } } + /** 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 once on entering 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. Edge-triggered to avoid a warning per evaluation round. + * 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 effectiveThreads, DynamicAllocationState.GroupLoad load) { + String groupName, int registeredThreads, DynamicAllocationState.GroupLoad load) { boolean planningBound = DynamicAllocationState.isPlanningBound( - effectiveThreads, + registeredThreads, load.getBusyThreads(), load.getServiceablePlanned(), load.getPendingTables()); - if (planningBound) { - if (planningBoundGroups.add(groupName)) { - 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, - effectiveThreads - load.getBusyThreads(), - load.getPendingTables(), - AmoroManagementConf.OPTIMIZER_MAX_PLANNING_PARALLELISM.key()); - } - } else { - planningBoundGroups.remove(groupName); + 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()); } } @@ -1101,9 +1163,10 @@ private void scaleIfNeeded( groupName, name -> new PendingRegistrations(BOOT_TIMEOUT_MS)); DynamicAllocationState state = scaleStates.computeIfAbsent(groupName, name -> new DynamicAllocationState()); - int effectiveThreads = getTotalQuota(groupName) + pending.pendingThreads(now); + int registeredThreads = getTotalQuota(groupName); + int effectiveThreads = registeredThreads + pending.pendingThreads(now); DynamicAllocationState.GroupLoad load = queue.collectDynamicAllocationLoad(); - warnOnPlanningBoundTransition(groupName, effectiveThreads, load); + warnOnPlanningBoundTransition(groupName, registeredThreads, load); int addInstances = state.computeScaleUp( effectiveThreads, @@ -1130,12 +1193,25 @@ private void scaleIfNeeded( .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) { - pending.failed(resource.getResourceId()); - LOG.warn("Dynamic allocation scale-out failed for group {}", groupName, 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/test/java/org/apache/amoro/server/TestOptimizerScaleKeeper.java b/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerScaleKeeper.java index dca047f178..8a57acb6c4 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerScaleKeeper.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerScaleKeeper.java @@ -202,6 +202,11 @@ public void testDraFloorSatisfiedInExecutorParallelismUnits() throws Interrupted 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"); } /** From e236b39622c0237c4286fcd1111dceb80990b12e Mon Sep 17 00:00:00 2001 From: Jiwon Park Date: Sun, 12 Jul 2026 19:09:21 +0900 Subject: [PATCH 08/10] [AMORO-4271] AIP-5 Phase 2: keep floor handling consistent across validation and scaling Two floor edge cases: - A floor unreachable in executor-parallelism units (e.g. min=5, max=6, K=4: covering the floor needs 2 instances = 8 threads > max) passed validation and then sat permanently below its floor with no signal - the same silent no-op the validation rules exist to prevent. Reject it up front; the keeper-side cap clamp stays as defense in depth for configs persisted before this rule. - A floor deficit (optimizers died, or a new group) now resets the demand-phase state: previously a demand phase before the deficit left a passed cadence gate and a grown ramp behind, so the first demand after recovery fired immediately and oversized instead of re-proving backlog persistence. Also pins that a fractional absolute quota truncates like the poll gate does ((int) 2.5 = 2 slots), keeping the two accountings aligned. Signed-off-by: Jiwon Park --- .../dra/DynamicAllocationConfig.java | 18 +++++++++++++++ .../dra/DynamicAllocationState.java | 6 +++++ .../optimizing/dra/TestComputeScaleUp.java | 18 +++++++++++++++ .../dra/TestDynamicAllocationConfig.java | 22 +++++++++++++++++++ .../dra/TestDynamicAllocationState.java | 10 +++++++++ 5 files changed, 74 insertions(+) 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 b9ec526465..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 @@ -299,6 +299,24 @@ public void validate() { 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); diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java index badb21f04d..9ac84334a3 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java @@ -65,6 +65,12 @@ public int computeScaleUp( 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); } diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestComputeScaleUp.java b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestComputeScaleUp.java index 08ecd543d7..d5b08ec566 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestComputeScaleUp.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestComputeScaleUp.java @@ -62,6 +62,8 @@ void floorDeficitScalesImmediately() { 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)); } @@ -151,6 +153,22 @@ void addIsCappedByMaxParallelismAndStopsAtCap() { 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(); 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 db7374a66c..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 @@ -102,6 +102,28 @@ void executorParallelismAboveMaxParallelismIsRejected() { 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 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 index b5837fa2ab..fc5233aee8 100644 --- 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 @@ -82,6 +82,16 @@ void absoluteQuotaOverOccupiedDoesNotGoNegative() { 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. From f250e4c6fa9fee9335a0c11bd24d2b9e3a83f45a Mon Sep 17 00:00:00 2001 From: Jiwon Park Date: Sun, 12 Jul 2026 19:09:30 +0900 Subject: [PATCH 09/10] [AMORO-4271] AIP-5 Phase 2: snapshot table runtimes under the scheduling lock SchedulingPolicy's table runtime map is a plain HashMap whose canonical accesses all hold tableLock; the dynamic-allocation load snapshot iterated it lock-free from the scale keeper thread, so a concurrent addTable/removeTable could throw ConcurrentModificationException and skip that whole evaluation round. Add a snapshot accessor that copies the runtimes under the lock and use it for the load snapshot. Also covers collectDynamicAllocationLoad with a queue-level test on a real table: the PENDING table is the only demand signal before any poll (the cold-start case), and a polled task occupies its thread from SCHEDULED on. Signed-off-by: Jiwon Park --- .../server/optimizing/OptimizingQueue.java | 2 +- .../server/optimizing/SchedulingPolicy.java | 16 ++++++++++++ .../optimizing/TestOptimizingQueue.java | 26 +++++++++++++++++++ 3 files changed, 43 insertions(+), 1 deletion(-) 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 b74562fb95..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 @@ -439,7 +439,7 @@ public DynamicAllocationState.GroupLoad collectDynamicAllocationLoad() { } Map targetQuotaByTable = Maps.newHashMap(); int pendingTables = 0; - for (DefaultTableRuntime tableRuntime : scheduler.getTableRuntimeMap().values()) { + for (DefaultTableRuntime tableRuntime : scheduler.snapshotTableRuntimes()) { targetQuotaByTable.put( tableRuntime.getTableIdentifier().getId(), tableRuntime.getOptimizingConfig().getTargetQuota()); 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/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java index 014189f64e..7bff3d6c78 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 @@ -48,6 +48,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; @@ -197,6 +198,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(); From 509f94a7746bb6a4bb2f9728ebc0114fdc56d11d Mon Sep 17 00:00:00 2001 From: Jiwon Park Date: Sun, 12 Jul 2026 19:09:40 +0900 Subject: [PATCH 10/10] [AMORO-4271] AIP-5 Phase 2: document the dynamic-allocation group properties Add the dynamic-allocation.* properties to the optimizer group property table, mark the flat min-parallelism row deprecated in favor of the namespaced key, and recommend executor-parallelism 4-8 for Kubernetes groups so per-pod JVM overhead is shared across threads. The scale-down-related properties are documented as landing in a later release. Signed-off-by: Jiwon Park --- docs/admin-guides/managing-optimizers.md | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) 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. |