[hotfix] Refactor autoscaler related setClock to optimize testing

Author: 1996fanrui

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

@@ -251,5 +251,7 @@ private AutoscalerFlinkMetrics getOrInitAutoscalerFlinkMetrics(Context ctx) {
     @VisibleForTesting
     void setClock(Clock clock) {
         this.clock = Preconditions.checkNotNull(clock);
+        this.metricsCollector.setClock(clock);
+        this.scalingExecutor.setClock(clock);
     }
 }

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

@@ -41,6 +41,7 @@
 
 import javax.annotation.Nullable;
 
+import java.time.Clock;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.Collections;
@@ -494,4 +495,9 @@ private boolean checkIfBlockedAndTriggerScalingEvent(
 
         return !scaleEnabled || isExcluded;
     }
+
+    @VisibleForTesting
+    void setClock(Clock clock) {
+        jobVertexScaler.setClock(clock);
+    }
 }

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

@@ -416,7 +416,6 @@ private void assertCollectedMetricsSize(int expectedSize) throws Exception {
 
     private void setClocksTo(Instant time) {
         var clock = Clock.fixed(time, ZoneId.systemDefault());
-        metricsCollector.setClock(clock);
         autoscaler.setClock(clock);
     }
 

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

@@ -278,7 +278,6 @@ private void running(Instant now) {
 
     private void setClocksTo(Instant time) {
         var clock = Clock.fixed(time, ZoneId.systemDefault());
-        metricsCollector.setClock(clock);
         autoscaler.setClock(clock);
     }
 }