This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new c305a7ce [FLINK-40389] Source parallelism is not capped at the
partition count (#1179)
c305a7ce is described below
commit c305a7ce554e8dd1a2381867824a260d43e70bbc
Author: Dennis-Mircea Ciupitu <[email protected]>
AuthorDate: Fri Aug 14 16:54:03 2026 +0300
[FLINK-40389] Source parallelism is not capped at the partition count
(#1179)
---
.../autoscaler/alignment/BuiltInAlignmentMode.java | 7 +-
.../autoscaler/alignment/ParallelismAligner.java | 5 ++
.../flink/autoscaler/JobVertexScalerTest.java | 43 ++++++++++
.../autoscaler/alignment/AlignmentModeTest.java | 99 ++++++++++++++++++++++
4 files changed, 153 insertions(+), 1 deletion(-)
diff --git
a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/BuiltInAlignmentMode.java
b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/BuiltInAlignmentMode.java
index 37dcc731..a85deab4 100644
---
a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/BuiltInAlignmentMode.java
+++
b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/BuiltInAlignmentMode.java
@@ -69,7 +69,12 @@ public enum BuiltInAlignmentMode implements
ParallelismAlignmentMode, DescribedE
*/
private static int alignOrKeepTarget(Context<?> ctx, boolean
acceptLoadReducing) {
int aligned = ParallelismAligner.firstAlignedInRegion(ctx,
acceptLoadReducing);
- return aligned > 0 ? aligned : ctx.getNewParallelism();
+ if (aligned > 0) {
+ return aligned;
+ }
+ // Nothing aligned in the region: keep the target, but never above the
cap, since
+ // subtasks past the source partition count are assigned no split.
+ return Math.min(ctx.getNewParallelism(),
ParallelismAligner.upperBoundForAlignment(ctx));
}
private final InlineElement description;
diff --git
a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/ParallelismAligner.java
b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/ParallelismAligner.java
index b58da779..65c02d68 100644
---
a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/ParallelismAligner.java
+++
b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/ParallelismAligner.java
@@ -280,6 +280,11 @@ public final class ParallelismAligner {
private static boolean invertsDirection(
ParallelismAlignmentMode.Context<?> ctx, int candidate) {
+ // Exception to the direction check below: the cap is the most
subtasks the vertex can
+ // use, so dropping onto it corrects an over-provisioned vertex rather
than inverting it.
+ if (candidate == upperBoundForAlignment(ctx) && candidate <
ctx.getCurrentParallelism()) {
+ return false;
+ }
return (isScaleUp(ctx) && candidate <= ctx.getCurrentParallelism())
|| (!isScaleUp(ctx) && candidate >=
ctx.getCurrentParallelism());
}
diff --git
a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobVertexScalerTest.java
b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobVertexScalerTest.java
index 8cba8221..eb7013b3 100644
---
a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobVertexScalerTest.java
+++
b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobVertexScalerTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.autoscaler;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.autoscaler.JobVertexScaler.ParallelismChange;
+import org.apache.flink.autoscaler.alignment.BuiltInAlignmentMode;
import org.apache.flink.autoscaler.alignment.KeyGroupOrPartitionsAdjustMode;
import org.apache.flink.autoscaler.config.AutoScalerOptions;
import org.apache.flink.autoscaler.event.TestingEventCollector;
@@ -1210,6 +1211,48 @@ public class JobVertexScalerTest {
context));
}
+ @Test
+ public void testSourceCappedAtPartitionCountWhenOverProvisioned() {
+ final var vertex = new JobVertexID();
+ // 15 partitions, running at 100: both a requested scale-up and
scale-down land on 15.
+ // setup() pins the legacy EVENLY_SPREAD, so that one is covered first.
+ assertEquals(15, scaleOverProvisionedSource(vertex, 1.8));
+ assertEquals(15, scaleOverProvisionedSource(vertex, 0.5));
+
+ conf.set(
+ AutoScalerOptions.SCALING_KEY_GROUP_PARTITIONS_ADJUST_MODE,
+ KeyGroupOrPartitionsAdjustMode.MAXIMIZE_UTILISATION);
+ assertEquals(15, scaleOverProvisionedSource(vertex, 1.8));
+ assertEquals(15, scaleOverProvisionedSource(vertex, 0.5));
+
+ // The new key takes precedence over the deprecated one set above.
+ for (var mode :
+ List.of(BuiltInAlignmentMode.BALANCED,
BuiltInAlignmentMode.EVENLY_SPREAD)) {
+ conf.set(AutoScalerOptions.ALIGNMENT_MODE, mode.name());
+ assertEquals(15, scaleOverProvisionedSource(vertex, 1.8),
mode.name());
+ assertEquals(15, scaleOverProvisionedSource(vertex, 0.5),
mode.name());
+ }
+
+ // OFF opts out of alignment, so the computed target is used as-is.
+ conf.set(AutoScalerOptions.ALIGNMENT_MODE,
BuiltInAlignmentMode.OFF.name());
+ assertEquals(180, scaleOverProvisionedSource(vertex, 1.8));
+ }
+
+ private int scaleOverProvisionedSource(JobVertexID vertex, double
scaleFactor) {
+ return vertexScaler.scale(
+ vertex,
+ 100,
+ List.of(),
+ 15,
+ 180,
+ scaleFactor,
+ 1,
+ Integer.MAX_VALUE,
+ Map.of(),
+ null,
+ context);
+ }
+
@Test
public void testSendingScalingLimitedEvents() {
var jobVertexID = new JobVertexID();
diff --git
a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/alignment/AlignmentModeTest.java
b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/alignment/AlignmentModeTest.java
index 90cd0f4b..33aa4223 100644
---
a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/alignment/AlignmentModeTest.java
+++
b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/alignment/AlignmentModeTest.java
@@ -23,14 +23,19 @@ import org.apache.flink.autoscaler.topology.ShipStrategy;
import org.apache.flink.configuration.Configuration;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
import static
org.apache.flink.autoscaler.TestingAutoscalerUtils.createDefaultJobAutoScalerContext;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
/**
* Tests for the built-in and legacy {@link ParallelismAlignmentMode}s and
{@link
@@ -123,6 +128,100 @@ class AlignmentModeTest {
assertThat(emitted).hasValue(1);
}
+ @SuppressWarnings("deprecation")
+ static Stream<ParallelismAlignmentMode> aligningModes() {
+ return Stream.of(
+ BuiltInAlignmentMode.BALANCED,
+ BuiltInAlignmentMode.EVENLY_SPREAD,
+ KeyGroupOrPartitionsAdjustMode.EVENLY_SPREAD,
+ KeyGroupOrPartitionsAdjustMode.MAXIMIZE_UTILISATION);
+ }
+
+ @SuppressWarnings("deprecation")
+ static Stream<KeyGroupOrPartitionsAdjustMode> legacyModes() {
+ return Stream.of(
+ KeyGroupOrPartitionsAdjustMode.EVENLY_SPREAD,
+ KeyGroupOrPartitionsAdjustMode.MAXIMIZE_UTILISATION);
+ }
+
+ /** Every aligning mode against a target above the cap, on both sides of
the current value. */
+ static Stream<Arguments> cappedTargets() {
+ return aligningModes()
+ .flatMap(mode -> Stream.of(180, 50, 16, 15).map(target ->
arguments(mode, target)));
+ }
+
+ @ParameterizedTest(name = "{0}, target {1}")
+ @MethodSource("cappedTargets")
+ void targetAbovePartitionCountIsCappedInEitherDirection(
+ ParallelismAlignmentMode mode, int target) {
+ // 15 partitions, running at 100: 15 is the ceiling whether the target
is above it (180),
+ // below it (50, 16), or already on it (15).
+ assertThat(mode.alignParallelism(ctx(100, target, 15,
180))).isEqualTo(15);
+ }
+
+ @Test
+ void offKeepsTheTargetAbovePartitionCount() {
+ // OFF opts out of alignment altogether, so the over-provisioning is
the user's choice.
+ assertThat(BuiltInAlignmentMode.OFF.alignParallelism(ctx(100, 180, 15,
180)))
+ .isEqualTo(180);
+ assertThat(BuiltInAlignmentMode.OFF.alignParallelism(ctx(100, 50, 15,
180))).isEqualTo(50);
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("aligningModes")
+ void keyedVertexIsCappedAtItsKeyGroupCount(ParallelismAlignmentMode mode) {
+ // Without source partitions the cap is the key group count. scale()
already clamps the
+ // target to maxParallelism, so only sources reach this in practice.
+ assertThat(mode.alignParallelism(ctx(20, 18, 0, 12))).isEqualTo(12);
+ }
+
+ /** Legacy modes only, and only targets that deviate from the cap, so an
event is expected. */
+ static Stream<Arguments> legacyCappedTargets() {
+ return legacyModes()
+ .flatMap(mode -> Stream.of(180, 50, 16).map(target ->
arguments(mode, target)));
+ }
+
+ @ParameterizedTest(name = "{0}, target {1}")
+ @MethodSource("legacyCappedTargets")
+ @SuppressWarnings("deprecation")
+ void legacyEmitsWheneverCappedAtPartitionCount(
+ KeyGroupOrPartitionsAdjustMode mode, int target) {
+ // Capping is not an inversion, but it still deviates from the target,
so the event stays.
+ assertThat(mode.alignParallelism(ctx(100, target, 15, 180),
emitter)).isEqualTo(15);
+ assertThat(emitted).hasValue(1);
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("legacyModes")
+ @SuppressWarnings("deprecation")
+ void
alreadyAtThePartitionCountBlocksAScaleUp(KeyGroupOrPartitionsAdjustMode mode) {
+ // Already on the cap: the candidate is not below current, so the
scale-up stays blocked.
+ assertThat(mode.alignParallelism(ctx(15, 20, 15, 180),
emitter)).isEqualTo(15);
+ assertThat(emitted).hasValue(1);
+ }
+
+ @Test
+ @SuppressWarnings("deprecation")
+ void movingUpToTheCapOnAScaleDownIsStillAnInversion() {
+ // The lower limit clamps the fallback up onto the cap (12), above the
current 10, so
+ // taking it would invert the scale-down.
+ var c = ctx(10, 8, 12, 128, 12, 12, List.of(ShipStrategy.HASH));
+
assertThat(KeyGroupOrPartitionsAdjustMode.EVENLY_SPREAD.alignParallelism(c,
emitter))
+ .isEqualTo(10);
+ assertThat(emitted).hasValue(1);
+ }
+
+ @Test
+ @SuppressWarnings("deprecation")
+ void lowerLimitClampAboveCurrentStillBlocksAScaleDown() {
+ // The same inversion, but below the cap (candidate 11, cap 30): the
exemption must not
+ // reach it.
+ var c = ctx(10, 8, 35, 128, 11, 30, List.of(ShipStrategy.HASH));
+
assertThat(KeyGroupOrPartitionsAdjustMode.EVENLY_SPREAD.alignParallelism(c,
emitter))
+ .isEqualTo(10);
+ assertThat(emitted).hasValue(1);
+ }
+
@Test
void modesDoNotApplyToNonSourceNonHashVertices() {
var rebalance = ctx(16, 24, 0, 128, 1, Integer.MAX_VALUE,
List.of(ShipStrategy.REBALANCE));
