Address Max's comments(third commit)

1996fanrui · Oct 12, 2023 · 3d0348b · 3d0348b
1 parent 4108026
commit 3d0348b
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 8 deletions.
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingHistoryUtils.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingHistoryUtils.java
@@ -89,17 +89,15 @@ Map<JobVertexID, SortedMap<Instant, ScalingSummary>> getTrimmedScalingHistory(
                 .getScalingHistory(context)
                 .map(
                         scalingHistory -> {
+                            Instant expectedStartTime =
+                                    now.minus(
+                                            conf.get(AutoScalerOptions.VERTEX_SCALING_HISTORY_AGE));
                             var entryIt = scalingHistory.entrySet().iterator();
                             while (entryIt.hasNext()) {
                                 var entry = entryIt.next();
                                 // Limit how long past scaling decisions are remembered
                                 entry.setValue(
-                                        entry.getValue()
-                                                .tailMap(
-                                                        now.minus(
-                                                                conf.get(
-                                                                        AutoScalerOptions
-                                                                                .VERTEX_SCALING_HISTORY_AGE))));
+                                        new TreeMap<>(entry.getValue().tailMap(expectedStartTime)));
                                 var vertexHistory = entry.getValue();
                                 while (vertexHistory.size()
                                         > conf.get(

diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/RestApiMetricsCollectorTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/RestApiMetricsCollectorTest.java
@@ -103,8 +103,6 @@ CompletableFuture<P> sendRequest(
         Map<JobVertexID, Map<FlinkMetric, AggregatedMetric>> jobVertexIDMapMap =
                 collector.queryAllAggregatedMetrics(context, metrics);
 
-        System.out.println(jobVertexIDMapMap);
-
         Assertions.assertEquals(1, jobVertexIDMapMap.size());
         Map<FlinkMetric, AggregatedMetric> vertexMetrics = jobVertexIDMapMap.get(jobVertexID);
         Assertions.assertNotNull(vertexMetrics);