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 d4e583dbe [FLINK-39743] [flink-autoscaler] Fix incorrect expected
processing rate computation
d4e583dbe is described below
commit d4e583dbe2bc334415af9eaa996f96268a0c4280
Author: Devika Sudheer <[email protected]>
AuthorDate: Tue Jun 16 22:11:01 2026 -0700
[FLINK-39743] [flink-autoscaler] Fix incorrect expected processing rate
computation
---
.../apache/flink/autoscaler/JobVertexScaler.java | 11 +++--
.../flink/autoscaler/JobVertexScalerTest.java | 56 ++++++++++++++++++++++
.../flink/autoscaler/ScalingExecutorTest.java | 2 +-
3 files changed, 63 insertions(+), 6 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 395a873d8..d634de261 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
@@ -226,10 +226,6 @@ public class JobVertexScaler<KEY, Context extends
JobAutoScalerContext<KEY>> {
scaleFactor = maxScaleFactor;
}
- // Cap target capacity according to the capped scale factor
- double cappedTargetCapacity = averageTrueProcessingRate * scaleFactor;
- LOG.debug("Capped target processing capacity for {} is {}", vertex,
cappedTargetCapacity);
-
int newParallelism =
scale(
vertex,
@@ -250,10 +246,15 @@ public class JobVertexScaler<KEY, Context extends
JobAutoScalerContext<KEY>> {
return ParallelismChange.noChange(currentParallelism);
}
+ // Cap expected target processing rate according to the capped scale
factor
+ double expectedProcessingRate =
+ (averageTrueProcessingRate / currentParallelism) *
newParallelism;
+ LOG.debug("Expected target processing rate for {} is {}", vertex,
expectedProcessingRate);
+
// We record our expectations for this scaling operation
evaluatedMetrics.put(
ScalingMetric.EXPECTED_PROCESSING_RATE,
- EvaluatedScalingMetric.of(cappedTargetCapacity));
+ EvaluatedScalingMetric.of(expectedProcessingRate));
return detectBlockScaling(
context,
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 22963c46f..2e7a43331 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
@@ -827,6 +827,62 @@ public class JobVertexScalerTest {
assertTrue(evaluated.containsKey(ScalingMetric.EXPECTED_PROCESSING_RATE));
}
+ @Test
+ public void testExpectedProcessingRateComputation() {
+ var jobVertexID = new JobVertexID();
+ conf.set(UTILIZATION_TARGET, 1.0);
+ conf.set(AutoScalerOptions.SCALE_DOWN_INTERVAL, Duration.ZERO);
+ var delayedScaleDown = new DelayedScaleDown();
+ var history = new TreeMap<Instant, ScalingSummary>();
+
+ // Max scale-up factor caps raw target parallelism 200 to 15
+ conf.set(AutoScalerOptions.MAX_SCALE_UP_FACTOR, 0.5);
+ var evaluated = evaluated(10, 200, 10);
+ assertEquals(
+ ParallelismChange.build(15, 10, true),
+ vertexScaler.computeScaleTargetParallelism(
+ context,
+ jobVertexID,
+ NOT_ADJUST_INPUTS,
+ evaluated,
+ history,
+ restartTime,
+ delayedScaleDown));
+ assertEquals(15.0,
evaluated.get(ScalingMetric.EXPECTED_PROCESSING_RATE).getCurrent());
+
+ // Key-group alignment rounds raw target parallelism 3.33 up to 4
+ conf.set(AutoScalerOptions.MAX_SCALE_UP_FACTOR, (double)
Integer.MAX_VALUE);
+ conf.set(UTILIZATION_TARGET, 0.6);
+ evaluated = evaluated(2, 100, 100);
+ assertEquals(
+ ParallelismChange.build(4, 2, true),
+ vertexScaler.computeScaleTargetParallelism(
+ context,
+ jobVertexID,
+ NOT_ADJUST_INPUTS,
+ evaluated,
+ history,
+ restartTime,
+ delayedScaleDown));
+ assertEquals(200.0,
evaluated.get(ScalingMetric.EXPECTED_PROCESSING_RATE).getCurrent());
+
+ // Source-partition count caps raw target parallelism 200 to 150
+ conf.set(UTILIZATION_TARGET, 1.0);
+ evaluated = evaluated(10, 200, 100);
+ evaluated.put(ScalingMetric.NUM_SOURCE_PARTITIONS,
EvaluatedScalingMetric.of(15));
+ assertEquals(
+ ParallelismChange.build(15, 10, true),
+ vertexScaler.computeScaleTargetParallelism(
+ context,
+ jobVertexID,
+ NOT_ADJUST_INPUTS,
+ evaluated,
+ history,
+ restartTime,
+ delayedScaleDown));
+ assertEquals(150.0,
evaluated.get(ScalingMetric.EXPECTED_PROCESSING_RATE).getCurrent());
+ }
+
@ParameterizedTest
@MethodSource("adjustmentInputsProvider")
public void testSendingIneffectiveScalingEvents(Collection<ShipStrategy>
inputShipStrategies) {
diff --git
a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java
b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java
index 1b1a10c14..ac07e1a32 100644
---
a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java
+++
b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java
@@ -724,7 +724,7 @@ public class ScalingExecutorTest {
1,
2,
100.0,
- 157.0,
+ 200.0,
110.0)));
assertTrue(
event.getMessage()
