[FLINK-36018][autoscaler] Support lazy scale down to avoid frequent r…

…escaling

flink-autoscaler-plugin-jdbc/src/main/java/org/apache/flink/autoscaler/jdbc/state/JdbcAutoScalerStateStore.java

@@ -19,6 +19,7 @@
 
 import org.apache.flink.annotation.Experimental;
 import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.autoscaler.DelayedScaleDown;
 import org.apache.flink.autoscaler.JobAutoScalerContext;
 import org.apache.flink.autoscaler.ScalingSummary;
 import org.apache.flink.autoscaler.ScalingTracking;
@@ -49,6 +50,7 @@
 
 import static org.apache.flink.autoscaler.jdbc.state.StateType.COLLECTED_METRICS;
 import static org.apache.flink.autoscaler.jdbc.state.StateType.CONFIG_OVERRIDES;
+import static org.apache.flink.autoscaler.jdbc.state.StateType.DELAYED_SCALE_DOWN;
 import static org.apache.flink.autoscaler.jdbc.state.StateType.PARALLELISM_OVERRIDES;
 import static org.apache.flink.autoscaler.jdbc.state.StateType.SCALING_HISTORY;
 import static org.apache.flink.autoscaler.jdbc.state.StateType.SCALING_TRACKING;
@@ -214,6 +216,35 @@ public void removeConfigChanges(Context jobContext) {
         jdbcStateStore.removeSerializedState(getSerializeKey(jobContext), CONFIG_OVERRIDES);
     }
 
+    @Override
+    public void storeDelayedScaleDown(Context jobContext, DelayedScaleDown delayedScaleDown)
+            throws Exception {
+        jdbcStateStore.putSerializedState(
+                getSerializeKey(jobContext),
+                DELAYED_SCALE_DOWN,
+                serializeDelayedScaleDown(delayedScaleDown));
+    }
+
+    @Nonnull
+    @Override
+    public DelayedScaleDown getDelayedScaleDown(Context jobContext) {
+        Optional<String> delayedScaleDown =
+                jdbcStateStore.getSerializedState(getSerializeKey(jobContext), DELAYED_SCALE_DOWN);
+        if (delayedScaleDown.isEmpty()) {
+            return new DelayedScaleDown();
+        }
+
+        try {
+            return deserializeDelayedScaleDown(delayedScaleDown.get());
+        } catch (JacksonException e) {
+            LOG.error(
+                    "Could not deserialize delayed scale down, possibly the format changed. Discarding...",
+                    e);
+            jdbcStateStore.removeSerializedState(getSerializeKey(jobContext), DELAYED_SCALE_DOWN);
+            return new DelayedScaleDown();
+        }
+    }
+
     @Override
     public void clearAll(Context jobContext) {
         jdbcStateStore.clearAll(getSerializeKey(jobContext));
@@ -296,4 +327,16 @@ private static ConfigChanges deserializeConfigOverrides(String configOverrides)
             return null;
         }
     }
+
+    private static String serializeDelayedScaleDown(DelayedScaleDown delayedScaleDown)
+            throws JacksonException {
+        return YAML_MAPPER.writeValueAsString(delayedScaleDown.getFirstTriggerTime());
+    }
+
+    private static DelayedScaleDown deserializeDelayedScaleDown(String delayedScaleDown)
+            throws JacksonException {
+        Map<JobVertexID, Instant> firstTriggerTime =
+                YAML_MAPPER.readValue(delayedScaleDown, new TypeReference<>() {});
+        return new DelayedScaleDown(firstTriggerTime);
+    }
 }

flink-autoscaler-plugin-jdbc/src/main/java/org/apache/flink/autoscaler/jdbc/state/StateType.java

@@ -27,7 +27,8 @@ public enum StateType {
     SCALING_TRACKING("scalingTracking"),
     COLLECTED_METRICS("collectedMetrics"),
     PARALLELISM_OVERRIDES("parallelismOverrides"),
-    CONFIG_OVERRIDES("configOverrides");
+    CONFIG_OVERRIDES("configOverrides"),
+    DELAYED_SCALE_DOWN("delayedScaleDown");
 
     /**
      * The identifier of each state type, it will be used to store. Please ensure the identifier is

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

@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.autoscaler;
+
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+
+import lombok.Getter;
+
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+/** All delayed scale down requests. */
+public class DelayedScaleDown {
+
+    @Getter private final Map<JobVertexID, Instant> firstTriggerTime;
+
+    // Have any scale down request been updated? It doesn't need to be stored, it is only used to
+    // determine whether DelayedScaleDown needs to be stored.
+    @Getter private boolean isUpdated = false;
+
+    public DelayedScaleDown() {
+        this.firstTriggerTime = new HashMap<>();
+    }
+
+    public DelayedScaleDown(Map<JobVertexID, Instant> firstTriggerTime) {
+        this.firstTriggerTime = firstTriggerTime;
+    }
+
+    Optional<Instant> getFirstTriggerTimeForVertex(JobVertexID vertex) {
+        return Optional.ofNullable(firstTriggerTime.get(vertex));
+    }
+
+    void updateTriggerTime(JobVertexID vertex, Instant instant) {
+        firstTriggerTime.put(vertex, instant);
+        isUpdated = true;
+    }
+
+    void clearVertex(JobVertexID vertex) {
+        Instant removed = firstTriggerTime.remove(vertex);
+        if (removed != null) {
+            isUpdated = true;
+        }
+    }
+
+    void clearAll() {
+        if (firstTriggerTime.isEmpty()) {
+            return;
+        }
+        firstTriggerTime.clear();
+        isUpdated = true;
+    }
+}

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

@@ -214,9 +214,20 @@ private void runScalingLogic(Context ctx, AutoscalerFlinkMetrics autoscalerMetri
             return;
         }
 
+        var delayedScaleDown = stateStore.getDelayedScaleDown(ctx);
         var parallelismChanged =
                 scalingExecutor.scaleResource(
-                        ctx, evaluatedMetrics, scalingHistory, scalingTracking, now, jobTopology);
+                        ctx,
+                        evaluatedMetrics,
+                        scalingHistory,
+                        scalingTracking,
+                        now,
+                        jobTopology,
+                        delayedScaleDown);
+
+        if (delayedScaleDown.isUpdated()) {
+            stateStore.storeDelayedScaleDown(ctx, delayedScaleDown);
+        }
 
         if (parallelismChanged) {
             autoscalerMetrics.incrementScaling();