fix test case

Author: huyuanfeng

flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetric.java

@@ -104,6 +104,7 @@ public enum ScalingMetric {
                     PARALLELISM,
                     RECOMMENDED_PARALLELISM,
                     MAX_PARALLELISM,
+                    NUM_PARTITIONS,
                     SCALE_UP_RATE_THRESHOLD,
                     SCALE_DOWN_RATE_THRESHOLD,
                     EXPECTED_PROCESSING_RATE);

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

@@ -233,21 +233,16 @@ private void setDefaultMetrics(
     }
 
     @Test
-    public void testKafkaPulsarPartitionMaxParallelism() throws Exception {
+    public void testKafkaPulsarNumPartitions() throws Exception {
         setDefaultMetrics(metricsCollector);
         metricsCollector.updateMetrics(context, stateStore);
 
         var clock = Clock.fixed(Instant.now().plus(Duration.ofSeconds(3)), ZoneId.systemDefault());
         metricsCollector.setClock(clock);
 
         var collectedMetrics = metricsCollector.updateMetrics(context, stateStore);
-
-        assertEquals(720, collectedMetrics.getJobTopology().get(source1).getMaxParallelism());
-        assertEquals(720, collectedMetrics.getJobTopology().get(source2).getMaxParallelism());
-
         clock = Clock.fixed(Instant.now().plus(Duration.ofSeconds(3)), ZoneId.systemDefault());
         metricsCollector.setClock(clock);
-
         metricsCollector.setMetricNames(
                 Map.of(
                         source1,
@@ -260,8 +255,7 @@ public void testKafkaPulsarPartitionMaxParallelism() throws Exception {
                                 "1.Source__Kafka_Source_(testTopic).KafkaSourceReader.topic.testTopic.partition.3.currentOffset")));
 
         collectedMetrics = metricsCollector.updateMetrics(context, stateStore);
-        assertEquals(5, collectedMetrics.getJobTopology().get(source1).getMaxParallelism());
-        assertEquals(720, collectedMetrics.getJobTopology().get(source2).getMaxParallelism());
+        assertEquals(5, collectedMetrics.getJobTopology().get(source1).getNumPartitions());
 
         metricsCollector.setMetricNames(
                 Map.of(
@@ -280,7 +274,7 @@ public void testKafkaPulsarPartitionMaxParallelism() throws Exception {
                                 "0.Source__pulsar_source[1].PulsarConsumer"
                                         + ".persistent_//public/default/testTopic-partition-4.m962n.numMsgsReceived")));
         collectedMetrics = metricsCollector.updateMetrics(context, stateStore);
-        assertEquals(5, collectedMetrics.getJobTopology().get(source2).getMaxParallelism());
+        assertEquals(5, collectedMetrics.getJobTopology().get(source2).getNumPartitions());
     }
 
     @Test

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

@@ -880,6 +880,7 @@ private Map<ScalingMetric, EvaluatedScalingMetric> evaluated(
         var metrics = new HashMap<ScalingMetric, EvaluatedScalingMetric>();
         metrics.put(ScalingMetric.PARALLELISM, EvaluatedScalingMetric.of(parallelism));
         metrics.put(ScalingMetric.MAX_PARALLELISM, EvaluatedScalingMetric.of(MAX_PARALLELISM));
+        metrics.put(ScalingMetric.NUM_PARTITIONS, EvaluatedScalingMetric.of(0));
         metrics.put(ScalingMetric.TARGET_DATA_RATE, new EvaluatedScalingMetric(target, target));
         metrics.put(ScalingMetric.CATCH_UP_DATA_RATE, EvaluatedScalingMetric.of(catchupRate));
         metrics.put(