1. Clamped lowerBound for scaling coefficient to 0.5

Author: pchoudhury22

flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java

@@ -187,7 +187,9 @@ public ParallelismChange computeScaleTargetParallelism(
         if (conf.get(OBSERVED_SCALABILITY_ENABLED)) {
 
             double scalingCoefficient =
-                    JobVertexScaler.calculateObservedScalingCoefficient(history, conf);
+                    JobVertexScaler.calculateObservedScalingCoefficient(
+                            history, conf.get(OBSERVED_SCALABILITY_MIN_OBSERVATIONS));
+
             scaleFactor = scaleFactor / scalingCoefficient;
         }
         double minScaleFactor = 1 - conf.get(MAX_SCALE_DOWN_FACTOR);
@@ -256,12 +258,14 @@ public ParallelismChange computeScaleTargetParallelism(
      * 1.0} is returned, assuming linear scaling.
      *
      * @param history A {@code SortedMap} of {@code Instant} timestamps to {@code ScalingSummary}
-     * @param conf Deployment configuration.
+     * @param minObservations The minimum number of observations required to compute the scaling
+     *     coefficient. If the number of historical entries is less than this threshold, a default
+     *     coefficient of {@code 1.0} is returned.
      * @return The computed scaling coefficient.
      */
     @VisibleForTesting
     protected static double calculateObservedScalingCoefficient(
-            SortedMap<Instant, ScalingSummary> history, Configuration conf) {
+            SortedMap<Instant, ScalingSummary> history, int minObservations) {
         /*
          * The scaling coefficient is computed using the least squares approach
          * to fit a linear model:
@@ -283,11 +287,8 @@ protected static double calculateObservedScalingCoefficient(
          *      α = ∑ (P_i * R_i) / (∑ (P_i^2) * β)
          *
          * We keep the system conservative for higher returns scenario by clamping computed α to an upper bound of 1.0.
-         * If the computed coefficient falls below threshold, the system falls back to assuming linear scaling (α = 1.0).
          */
 
-        var minObservations = conf.get(OBSERVED_SCALABILITY_MIN_OBSERVATIONS);
-
         // not enough data to compute scaling coefficient; we assume linear scaling.
         if (history.isEmpty() || history.size() < minObservations) {
             return 1.0;
@@ -319,17 +320,7 @@ protected static double calculateObservedScalingCoefficient(
 
         var coefficient =
                 AutoScalerUtils.optimizeLinearScalingCoefficient(
-                        parallelismList, processingRateList, baselineProcessingRate, 1, 0.01);
-
-        double threshold =
-                conf.get(AutoScalerOptions.SCALING_EFFECTIVENESS_DETECTION_ENABLED)
-                        ? conf.get(AutoScalerOptions.SCALING_EFFECTIVENESS_THRESHOLD)
-                        : 0.5;
-
-        if (coefficient < threshold) {
-            LOG.warn("Scaling coefficient is below threshold. Falling back to linear scaling.");
-            return 1.0;
-        }
+                        parallelismList, processingRateList, baselineProcessingRate, 1, 0.5);
 
         return BigDecimal.valueOf(coefficient).setScale(2, RoundingMode.CEILING).doubleValue();
     }

flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobVertexScalerTest.java

@@ -52,7 +52,6 @@
 import static org.apache.flink.autoscaler.config.AutoScalerOptions.OBSERVED_SCALABILITY_ENABLED;
 import static org.apache.flink.autoscaler.config.AutoScalerOptions.OBSERVED_SCALABILITY_MIN_OBSERVATIONS;
 import static org.apache.flink.autoscaler.config.AutoScalerOptions.UTILIZATION_TARGET;
-
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -1179,7 +1178,7 @@ public void testCalculateScalingCoefficient() {
                 currentTime, new ScalingSummary(8, 16, linearScalingEvaluatedData3));
 
         double linearScalingScalingCoefficient =
-                JobVertexScaler.calculateObservedScalingCoefficient(linearScalingHistory, conf);
+                JobVertexScaler.calculateObservedScalingCoefficient(linearScalingHistory, 3);
 
         assertEquals(1.0, linearScalingScalingCoefficient);
 
@@ -1199,7 +1198,7 @@ public void testCalculateScalingCoefficient() {
 
         double slightDiminishingReturnsScalingCoefficient =
                 JobVertexScaler.calculateObservedScalingCoefficient(
-                        slightDiminishingReturnsScalingHistory, conf);
+                        slightDiminishingReturnsScalingHistory, 3);
 
         assertTrue(
                 slightDiminishingReturnsScalingCoefficient > 0.9
@@ -1221,7 +1220,7 @@ public void testCalculateScalingCoefficient() {
 
         double sharpDiminishingReturnsScalingCoefficient =
                 JobVertexScaler.calculateObservedScalingCoefficient(
-                        sharpDiminishingReturnsScalingHistory, conf);
+                        sharpDiminishingReturnsScalingHistory, 3);
 
         assertTrue(
                 sharpDiminishingReturnsScalingCoefficient < 0.9
@@ -1245,7 +1244,7 @@ public void testCalculateScalingCoefficient() {
 
         double sharpDiminishingReturnsWithOneParallelismScalingCoefficient =
                 JobVertexScaler.calculateObservedScalingCoefficient(
-                        sharpDiminishingReturnsWithOneParallelismScalingHistory, conf);
+                        sharpDiminishingReturnsWithOneParallelismScalingHistory, 3);
 
         assertTrue(
                 sharpDiminishingReturnsWithOneParallelismScalingCoefficient < 0.9
@@ -1261,8 +1260,7 @@ public void testCalculateScalingCoefficient() {
                 currentTime, new ScalingSummary(4, 8, withOneScalingHistoryRecordEvaluatedData1));
 
         double withOneScalingHistoryRecordScalingCoefficient =
-                JobVertexScaler.calculateObservedScalingCoefficient(
-                        withOneScalingHistoryRecord, conf);
+                JobVertexScaler.calculateObservedScalingCoefficient(withOneScalingHistoryRecord, 1);
 
         assertEquals(1, withOneScalingHistoryRecordScalingCoefficient);
 
@@ -1282,7 +1280,7 @@ public void testCalculateScalingCoefficient() {
 
         double diminishingReturnWithTwoScalingHistoryRecordScalingCoefficient =
                 JobVertexScaler.calculateObservedScalingCoefficient(
-                        diminishingReturnWithTwoScalingHistoryRecord, conf);
+                        diminishingReturnWithTwoScalingHistoryRecord, 1);
 
         assertTrue(
                 diminishingReturnWithTwoScalingHistoryRecordScalingCoefficient < 0.9
@@ -1302,7 +1300,7 @@ public void testCalculateScalingCoefficient() {
 
         double linearReturnWithTwoScalingHistoryRecordScalingCoefficient =
                 JobVertexScaler.calculateObservedScalingCoefficient(
-                        linearReturnWithTwoScalingHistoryRecord, conf);
+                        linearReturnWithTwoScalingHistoryRecord, 1);
 
         assertEquals(1, linearReturnWithTwoScalingHistoryRecordScalingCoefficient);
     }