Address Max and yuanfeng's comments

Author: 1996fanrui

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

@@ -64,19 +64,19 @@ public DelayedScaleDown() {
     @Nonnull
     public VertexDelayedScaleDownInfo triggerScaleDown(
             JobVertexID vertex, Instant triggerTime, int parallelism) {
-        var vertexInfo = delayedVertices.get(vertex);
-        if (vertexInfo == null) {
+        var vertexDelayedScaleDownInfo = delayedVertices.get(vertex);
+        if (vertexDelayedScaleDownInfo == null) {
             // It's the first trigger
-            vertexInfo = new VertexDelayedScaleDownInfo(triggerTime, parallelism);
-            delayedVertices.put(vertex, vertexInfo);
+            vertexDelayedScaleDownInfo = new VertexDelayedScaleDownInfo(triggerTime, parallelism);
+            delayedVertices.put(vertex, vertexDelayedScaleDownInfo);
             updated = true;
-        } else if (parallelism > vertexInfo.getMaxRecommendedParallelism()) {
+        } else if (parallelism > vertexDelayedScaleDownInfo.getMaxRecommendedParallelism()) {
             // Not the first trigger, but the maxRecommendedParallelism needs to be updated.
-            vertexInfo.setMaxRecommendedParallelism(parallelism);
+            vertexDelayedScaleDownInfo.setMaxRecommendedParallelism(parallelism);
             updated = true;
         }
 
-        return vertexInfo;
+        return vertexDelayedScaleDownInfo;
     }
 
     // Clear the delayed scale down for corresponding vertex when the recommended parallelism is

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

@@ -181,12 +181,12 @@ private void updateRecommendedParallelism(
     static boolean allChangedVerticesWithinUtilizationTarget(
             Map<JobVertexID, Map<ScalingMetric, EvaluatedScalingMetric>> evaluatedMetrics,
             Set<JobVertexID> changedVertices) {
-        // No any vertex is changed.
+        // No vertices with changed parallelism.
         if (changedVertices.isEmpty()) {
             return true;
         }
 
-        for (var vertex : changedVertices) {
+        for (JobVertexID vertex : changedVertices) {
             var metrics = evaluatedMetrics.get(vertex);
 
             double trueProcessingRate = metrics.get(TRUE_PROCESSING_RATE).getAverage();