[FLINK-36863][autoscaler] Use the maximum parallelism in the past scale-down.interval window when scaling down

Author: 1996fanrui

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

@@ -30,23 +30,78 @@
 
 import java.time.Instant;
 import java.util.HashMap;
+import java.util.LinkedList;
 import java.util.Map;
 
+import static org.apache.flink.util.Preconditions.checkState;
+
 /** All delayed scale down requests. */
 public class DelayedScaleDown {
 
+    @Data
+    private static class RecommendedParallelism {
+        @Nonnull private final Instant triggerTime;
+        private final int parallelism;
+
+        @JsonCreator
+        public RecommendedParallelism(
+                @Nonnull @JsonProperty("triggerTime") Instant triggerTime,
+                @JsonProperty("parallelism") int parallelism) {
+            this.triggerTime = triggerTime;
+            this.parallelism = parallelism;
+        }
+    }
+
     /** The delayed scale down info for vertex. */
     @Data
     public static class VertexDelayedScaleDownInfo {
         private final Instant firstTriggerTime;
-        private int maxRecommendedParallelism;
+        // TODO : add the comment to explain how to calculate the max parallelism within the sliding
+        // window.
+        private final LinkedList<RecommendedParallelism> recommendedParallelisms;
+
+        public VertexDelayedScaleDownInfo(Instant firstTriggerTime) {
+            this.firstTriggerTime = firstTriggerTime;
+            this.recommendedParallelisms = new LinkedList<>();
+        }
 
         @JsonCreator
         public VertexDelayedScaleDownInfo(
                 @JsonProperty("firstTriggerTime") Instant firstTriggerTime,
-                @JsonProperty("maxRecommendedParallelism") int maxRecommendedParallelism) {
+                @JsonProperty("recommendedParallelisms")
+                        LinkedList<RecommendedParallelism> recommendedParallelisms) {
             this.firstTriggerTime = firstTriggerTime;
-            this.maxRecommendedParallelism = maxRecommendedParallelism;
+            this.recommendedParallelisms = recommendedParallelisms;
+        }
+
+        /** Record current recommended parallelism. */
+        public void recordRecommendedParallelism(Instant triggerTime, int parallelism) {
+
+            // Remove all recommended parallelisms that are lower than the latest parallelism.
+            while (!recommendedParallelisms.isEmpty()
+                    && recommendedParallelisms.peekLast().getParallelism() <= parallelism) {
+                recommendedParallelisms.pollLast();
+            }
+
+            recommendedParallelisms.addLast(new RecommendedParallelism(triggerTime, parallelism));
+        }
+
+        @JsonIgnore
+        public int getMaxRecommendedParallelism(Instant windowStartTime) {
+            // Remove all recommended parallelisms before the window start time.
+            while (!recommendedParallelisms.isEmpty()
+                    && recommendedParallelisms
+                            .peekFirst()
+                            .getTriggerTime()
+                            .isBefore(windowStartTime)) {
+                recommendedParallelisms.pollFirst();
+            }
+
+            var maxRecommendedParallelism = recommendedParallelisms.peekFirst();
+            checkState(
+                    maxRecommendedParallelism != null,
+                    "The getMaxRecommendedParallelism should be called after triggering a scale down, it may be a bug.");
+            return maxRecommendedParallelism.getParallelism();
         }
     }
 
@@ -64,17 +119,16 @@ public DelayedScaleDown() {
     @Nonnull
     public VertexDelayedScaleDownInfo triggerScaleDown(
             JobVertexID vertex, Instant triggerTime, int parallelism) {
+        // The vertexDelayedScaleDownInfo is updated once scale down is triggered due to we need
+        // update the triggerTime each time.
+        updated = true;
+
         var vertexDelayedScaleDownInfo = delayedVertices.get(vertex);
         if (vertexDelayedScaleDownInfo == null) {
-            // It's the first trigger
-            vertexDelayedScaleDownInfo = new VertexDelayedScaleDownInfo(triggerTime, parallelism);
+            vertexDelayedScaleDownInfo = new VertexDelayedScaleDownInfo(triggerTime);
             delayedVertices.put(vertex, vertexDelayedScaleDownInfo);
-            updated = true;
-        } else if (parallelism > vertexDelayedScaleDownInfo.getMaxRecommendedParallelism()) {
-            // Not the first trigger, but the maxRecommendedParallelism needs to be updated.
-            vertexDelayedScaleDownInfo.setMaxRecommendedParallelism(parallelism);
-            updated = true;
         }
+        vertexDelayedScaleDownInfo.recordRecommendedParallelism(triggerTime, parallelism);
 
         return vertexDelayedScaleDownInfo;
     }

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

@@ -278,10 +278,11 @@ private ParallelismChange applyScaleDownInterval(
         }
 
         var now = clock.instant();
+        var windowStartTime = now.minus(scaleDownInterval);
         var delayedScaleDownInfo = delayedScaleDown.triggerScaleDown(vertex, now, newParallelism);
 
         // Never scale down within scale down interval
-        if (now.isBefore(delayedScaleDownInfo.getFirstTriggerTime().plus(scaleDownInterval))) {
+        if (windowStartTime.isBefore(delayedScaleDownInfo.getFirstTriggerTime())) {
             if (now.equals(delayedScaleDownInfo.getFirstTriggerTime())) {
                 LOG.info("The scale down of {} is delayed by {}.", vertex, scaleDownInterval);
             } else {
@@ -293,7 +294,8 @@ private ParallelismChange applyScaleDownInterval(
         } else {
             // Using the maximum parallelism within the scale down interval window instead of the
             // latest parallelism when scaling down
-            return ParallelismChange.build(delayedScaleDownInfo.getMaxRecommendedParallelism());
+            return ParallelismChange.build(
+                    delayedScaleDownInfo.getMaxRecommendedParallelism(windowStartTime));
         }
     }
 

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

@@ -38,40 +38,49 @@ void testTriggerUpdateAndClean() {
 
         // First trigger time as the trigger time, and it won't be updated.
         assertVertexDelayedScaleDownInfo(
-                delayedScaleDown.triggerScaleDown(vertex, instant, 5), instant, 5);
+                delayedScaleDown.triggerScaleDown(vertex, instant, 5), instant, 5, instant);
         assertThat(delayedScaleDown.isUpdated()).isTrue();
 
         // The lower parallelism doesn't update the result
         assertVertexDelayedScaleDownInfo(
-                delayedScaleDown.triggerScaleDown(vertex, instant.plusSeconds(5), 3), instant, 5);
+                delayedScaleDown.triggerScaleDown(vertex, instant.plusSeconds(5), 3),
+                instant,
+                5,
+                instant);
 
         // The higher parallelism will update the result
         assertVertexDelayedScaleDownInfo(
-                delayedScaleDown.triggerScaleDown(vertex, instant.plusSeconds(10), 8), instant, 8);
+                delayedScaleDown.triggerScaleDown(vertex, instant.plusSeconds(10), 8),
+                instant,
+                8,
+                instant);
 
         // The scale down could be re-triggered again after clean
         delayedScaleDown.clearVertex(vertex);
         assertThat(delayedScaleDown.getDelayedVertices()).isEmpty();
         assertVertexDelayedScaleDownInfo(
                 delayedScaleDown.triggerScaleDown(vertex, instant.plusSeconds(15), 4),
                 instant.plusSeconds(15),
-                4);
+                4,
+                instant);
 
         // The scale down could be re-triggered again after cleanAll
         delayedScaleDown.clearAll();
         assertThat(delayedScaleDown.getDelayedVertices()).isEmpty();
         assertVertexDelayedScaleDownInfo(
                 delayedScaleDown.triggerScaleDown(vertex, instant.plusSeconds(15), 2),
                 instant.plusSeconds(15),
-                2);
+                2,
+                instant);
     }
 
     void assertVertexDelayedScaleDownInfo(
             DelayedScaleDown.VertexDelayedScaleDownInfo vertexDelayedScaleDownInfo,
             Instant expectedTriggerTime,
-            int expectedMaxRecommendedParallelism) {
+            int expectedMaxRecommendedParallelism,
+            Instant windowStartTime) {
         assertThat(vertexDelayedScaleDownInfo.getFirstTriggerTime()).isEqualTo(expectedTriggerTime);
-        assertThat(vertexDelayedScaleDownInfo.getMaxRecommendedParallelism())
+        assertThat(vertexDelayedScaleDownInfo.getMaxRecommendedParallelism(windowStartTime))
                 .isEqualTo(expectedMaxRecommendedParallelism);
     }
 }