You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2026/04/02 13:15:13 UTC
(flink-kubernetes-operator) branch main updated: [FLINK-39299][flink-autoscaler] Inconsistent vertex parallelism alignment logic
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 14f04033 [FLINK-39299][flink-autoscaler] Inconsistent vertex parallelism alignment logic
14f04033 is described below
commit 14f04033fc18fbde91bca14b6d38bbdee4ad3722
Author: Dennis-Mircea Ciupitu <de...@gmail.com>
AuthorDate: Thu Apr 2 16:15:06 2026 +0300
[FLINK-39299][flink-autoscaler] Inconsistent vertex parallelism alignment logic
---
.../apache/flink/autoscaler/JobVertexScaler.java | 115 +++++++++++++++++----
.../flink/autoscaler/JobAutoScalerImplTest.java | 6 +-
.../flink/autoscaler/JobVertexScalerTest.java | 6 +-
3 files changed, 103 insertions(+), 24 deletions(-)
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java
index 492615f4..dfad27ad 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java
@@ -221,8 +221,8 @@ public class JobVertexScaler<KEY, Context extends JobAutoScalerContext<KEY>> {
(int) evaluatedMetrics.get(NUM_SOURCE_PARTITIONS).getCurrent(),
(int) evaluatedMetrics.get(MAX_PARALLELISM).getCurrent(),
scaleFactor,
- Math.min(currentParallelism, conf.getInteger(VERTEX_MIN_PARALLELISM)),
- Math.max(currentParallelism, conf.getInteger(VERTEX_MAX_PARALLELISM)),
+ Math.min(currentParallelism, conf.get(VERTEX_MIN_PARALLELISM)),
+ Math.max(currentParallelism, conf.get(VERTEX_MAX_PARALLELISM)),
autoScalerEventHandler,
context);
@@ -483,16 +483,28 @@ public class JobVertexScaler<KEY, Context extends JobAutoScalerContext<KEY>> {
}
/**
- * Computing the newParallelism. In general, newParallelism = currentParallelism * scaleFactor.
- * But we limit newParallelism between parallelismLowerLimit and min(parallelismUpperLimit,
- * maxParallelism).
+ * Computes the new parallelism for a vertex. In general, {@code newParallelism =
+ * currentParallelism * scaleFactor}, clamped to {@code [parallelismLowerLimit,
+ * min(parallelismUpperLimit, maxParallelism)]}.
*
- * <p>Also, in order to ensure the data is evenly spread across subtasks, we try to adjust the
- * parallelism for source and keyed vertex such that it divides the maxParallelism without a
- * remainder.
+ * <p>For source vertices (with known partition counts) and keyed vertices (with HASH inputs),
+ * the parallelism is further adjusted to align with the number of key groups or source
+ * partitions, according to the configured {@link KeyGroupOrPartitionsAdjustMode}.
*
- * <p>This method also attempts to adjust the parallelism to ensure it aligns well with the
- * number of source partitions if a vertex has a known source partition count.
+ * <p>The alignment uses a two-phase algorithm:
+ *
+ * <ol>
+ * <li><b>Phase 1 (upward search):</b> searches from the computed {@code newParallelism}
+ * upward for divisor-aligned values. For scale-down, the search is capped at {@code
+ * currentParallelism} to prevent the alignment from crossing into the scale-up direction.
+ * If the nearest divisor is {@code currentParallelism} itself, it is returned
+ * (effectively blocking the scale-down).
+ * <li><b>Phase 2 (downward fallback):</b> if Phase 1 finds no match, searches downward from
+ * {@code newParallelism} for the boundary where per-subtask load increases. A
+ * direction-safety guard ensures the result never inverts the intended scaling direction
+ * (scale-up results stay above {@code currentParallelism}, scale-down results stay
+ * below).
+ * </ol>
*/
@VisibleForTesting
protected static <KEY, Context extends JobAutoScalerContext<KEY>> int scale(
@@ -548,10 +560,23 @@ public class JobVertexScaler<KEY, Context extends JobAutoScalerContext<KEY>> {
KeyGroupOrPartitionsAdjustMode mode =
context.getConfiguration().get(SCALING_KEY_GROUP_PARTITIONS_ADJUST_MODE);
- // When the shuffle type of vertex inputs contains keyBy or vertex is a source,
- // we try to adjust the parallelism such that it divides
- // the numKeyGroupsOrPartitions without a remainder => data is evenly spread across subtasks
- for (int p = newParallelism; p <= upperBoundForAlignment; p++) {
+ if (newParallelism == currentParallelism) {
+ return newParallelism;
+ }
+
+ boolean isScaleUp = newParallelism > currentParallelism;
+
+ // Phase 1: Upward search from newParallelism for divisor or approximate match.
+ // For scale-down, cap search at currentParallelism to prevent the alignment
+ // from crossing into the scale-up direction (direction-safety fix).
+ // Returning currentParallelism itself is allowed — it effectively blocks the
+ // scale-down since no aligned value was found strictly below it.
+ int phase1UpperBound =
+ isScaleUp
+ ? upperBoundForAlignment
+ : Math.min(currentParallelism, upperBoundForAlignment);
+
+ for (int p = newParallelism; p <= phase1UpperBound; p++) {
if (numKeyGroupsOrPartitions % p == 0
||
// When Mode is MAXIMIZE_UTILISATION , Try to find the smallest parallelism
@@ -575,14 +600,67 @@ public class JobVertexScaler<KEY, Context extends JobAutoScalerContext<KEY>> {
break;
}
}
-
p = Math.max(p, parallelismLowerLimit);
+
+ // Direction-safety guard: prevent the alignment from inverting the scaling direction.
+ if ((isScaleUp && p <= currentParallelism) || (!isScaleUp && p >= currentParallelism)) {
+ LOG.warn(
+ "Scaling of {} blocked: alignment could not find a value that preserves the "
+ + "{} direction from current parallelism {} with mode {}. "
+ + "Keeping current parallelism.",
+ vertex,
+ isScaleUp ? "scale-up" : "scale-down",
+ currentParallelism,
+ mode);
+ emitScalingLimitedEvent(
+ vertex,
+ newParallelism,
+ currentParallelism,
+ numKeyGroupsOrPartitions,
+ upperBound,
+ parallelismLowerLimit,
+ eventHandler,
+ context);
+ return currentParallelism;
+ }
+
+ if (p != newParallelism) {
+ emitScalingLimitedEvent(
+ vertex,
+ newParallelism,
+ p,
+ numKeyGroupsOrPartitions,
+ upperBound,
+ parallelismLowerLimit,
+ eventHandler,
+ context);
+ }
+ return p;
+ }
+
+ /**
+ * Emits a {@link #SCALING_LIMITED} warning event when the alignment logic had to deviate from
+ * the computed target parallelism, or when no aligned parallelism could be found and the
+ * scaling operation was blocked.
+ *
+ * @param expectedParallelism the originally computed target parallelism
+ * @param actualParallelism the parallelism that will actually be applied
+ */
+ private static <KEY, Context extends JobAutoScalerContext<KEY>> void emitScalingLimitedEvent(
+ JobVertexID vertex,
+ int expectedParallelism,
+ int actualParallelism,
+ int numKeyGroupsOrPartitions,
+ int upperBound,
+ int parallelismLowerLimit,
+ AutoScalerEventHandler<KEY, Context> eventHandler,
+ Context context) {
var message =
String.format(
SCALE_LIMITED_MESSAGE_FORMAT,
vertex,
- newParallelism,
- p,
+ expectedParallelism,
+ actualParallelism,
numKeyGroupsOrPartitions,
upperBound,
parallelismLowerLimit);
@@ -591,9 +669,8 @@ public class JobVertexScaler<KEY, Context extends JobAutoScalerContext<KEY>> {
AutoScalerEventHandler.Type.Warning,
SCALING_LIMITED,
message,
- SCALING_LIMITED + vertex + newParallelism,
+ SCALING_LIMITED + vertex,
context.getConfiguration().get(SCALING_EVENT_INTERVAL));
- return p;
}
@VisibleForTesting
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java
index 552971d7..7ba9e88c 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java
@@ -201,11 +201,13 @@ public class JobAutoScalerImplTest {
new ScalingRealizer<>() {
@Override
public void realizeConfigOverrides(
- JobAutoScalerContext context, ConfigChanges configChanges) {}
+ JobAutoScalerContext<JobID> context,
+ ConfigChanges configChanges) {}
@Override
public void realizeParallelismOverrides(
- JobAutoScalerContext context, Map parallelismOverrides) {
+ JobAutoScalerContext<JobID> context,
+ Map<String, String> parallelismOverrides) {
throw new RuntimeException(
"Test Realize Parallelism Overrides Exceptions.");
}
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 3d085e17..c8ec9214 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
@@ -530,7 +530,7 @@ public class JobVertexScalerTest {
@Test
public void testMinParallelismLimitIsUsed() {
- conf.setInteger(AutoScalerOptions.VERTEX_MIN_PARALLELISM, 5);
+ conf.set(AutoScalerOptions.VERTEX_MIN_PARALLELISM, 5);
conf.set(AutoScalerOptions.SCALE_DOWN_INTERVAL, Duration.ZERO);
var delayedScaleDown = new DelayedScaleDown();
@@ -560,7 +560,7 @@ public class JobVertexScalerTest {
@Test
public void testMaxParallelismLimitIsUsed() {
- conf.setInteger(AutoScalerOptions.VERTEX_MAX_PARALLELISM, 10);
+ conf.set(AutoScalerOptions.VERTEX_MAX_PARALLELISM, 10);
conf.set(UTILIZATION_TARGET, 1.);
var delayedScaleDown = new DelayedScaleDown();
@@ -1015,7 +1015,7 @@ public class JobVertexScalerTest {
context));
assertEquals(
- 20,
+ 22,
JobVertexScaler.scale(
vertex,
22,