[FLINK-35328] Autoscaler supports setting the maximum floating parallelism by the number of Pulsar partitions

Author: wenbingshen

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

@@ -58,12 +58,14 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.SortedMap;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
+import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -202,7 +204,7 @@ protected JobTopology getJobTopology(
 
         Set<JobVertexID> vertexSet = Set.copyOf(t.getVerticesInTopologicalOrder());
         updateVertexList(stateStore, ctx, clock.instant(), vertexSet);
-        updateKafkaSourceMaxParallelisms(ctx, jobDetailsInfo.getJobId(), t);
+        updateKafkaPulsarSourceMaxParallelisms(ctx, jobDetailsInfo.getJobId(), t);
         excludeVerticesFromScaling(ctx.getConfiguration(), t.getFinishedVertices());
         return t;
     }
@@ -247,17 +249,36 @@ protected JobTopology getJobTopology(JobDetailsInfo jobDetailsInfo) {
                 json, slotSharingGroupIdMap, maxParallelismMap, metrics, finished);
     }
 
-    private void updateKafkaSourceMaxParallelisms(Context ctx, JobID jobId, JobTopology topology)
-            throws Exception {
+    private void updateKafkaPulsarSourceMaxParallelisms(
+            Context ctx, JobID jobId, JobTopology topology) throws Exception {
         try (var restClient = ctx.getRestClusterClient()) {
-            var partitionRegex = Pattern.compile("^.*\\.partition\\.\\d+\\.currentOffset$");
+            Pattern partitionRegex =
+                    Pattern.compile(
+                            "^.*\\.KafkaSourceReader\\.topic\\.(?<kafkaTopic>.+)\\.partition\\.(?<kafkaId>\\d+)\\.currentOffset$"
+                                    + "|^.*\\.PulsarConsumer\\.(?<pulsarTopic>.+)-partition-(?<pulsarId>\\d+)\\..*\\.numMsgsReceived$");
             for (var vertexInfo : topology.getVertexInfos().values()) {
                 if (vertexInfo.getInputs().isEmpty()) {
                     var sourceVertex = vertexInfo.getId();
                     var numPartitions =
                             queryAggregatedMetricNames(restClient, jobId, sourceVertex).stream()
-                                    .filter(partitionRegex.asMatchPredicate())
-                                    .count();
+                                    .map(
+                                            v -> {
+                                                Matcher matcher = partitionRegex.matcher(v);
+                                                if (matcher.matches()) {
+                                                    String kafkaTopic = matcher.group("kafkaTopic");
+                                                    String kafkaId = matcher.group("kafkaId");
+                                                    String pulsarTopic =
+                                                            matcher.group("pulsarTopic");
+                                                    String pulsarId = matcher.group("pulsarId");
+                                                    return kafkaTopic != null
+                                                            ? kafkaTopic + "-" + kafkaId
+                                                            : pulsarTopic + "-" + pulsarId;
+                                                }
+                                                return null;
+                                            })
+                                    .filter(Objects::nonNull)
+                                    .collect(Collectors.toSet())
+                                    .size();
                     if (numPartitions > 0) {
                         LOG.debug(
                                 "Updating source {} max parallelism based on available partitions to {}",

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

@@ -232,7 +232,7 @@ private void setDefaultMetrics(
     }
 
     @Test
-    public void testKafkaPartitionMaxParallelism() throws Exception {
+    public void testKafkaPulsarPartitionMaxParallelism() throws Exception {
         setDefaultMetrics(metricsCollector);
         metricsCollector.updateMetrics(context, stateStore);
 
@@ -261,6 +261,25 @@ public void testKafkaPartitionMaxParallelism() throws Exception {
         collectedMetrics = metricsCollector.updateMetrics(context, stateStore);
         assertEquals(5, collectedMetrics.getJobTopology().get(source1).getMaxParallelism());
         assertEquals(720, collectedMetrics.getJobTopology().get(source2).getMaxParallelism());
+
+        metricsCollector.setMetricNames(
+                Map.of(
+                        source2,
+                        List.of(
+                                "0.Source__pulsar_source[1].PulsarConsumer"
+                                        + ".persistent_//public/default/testTopic-partition-1.d842f.numMsgsReceived",
+                                "0.Source__pulsar_source[1].PulsarConsumer"
+                                        + ".persistent_//public/default/testTopic-partition-1.660d2.numMsgsReceived",
+                                "0.Source__pulsar_source[1].PulsarConsumer"
+                                        + ".persistent_//public/default/testTopic-partition-2.d356f.numMsgsReceived",
+                                "0.Source__pulsar_source[1].PulsarConsumer"
+                                        + ".persistent_//public/default/otherTopic-partition-2.m953d.numMsgsReceived",
+                                "0.Source__pulsar_source[1].PulsarConsumer"
+                                        + ".persistent_//public/default/testTopic-partition-3.e427h.numMsgsReceived",
+                                "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());
     }
 
     @Test