You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2024/06/11 06:09:35 UTC
(flink-kubernetes-operator) branch main updated: [FLINK-35328] Autoscaler supports setting the maximum floating parallelism by the number of Pulsar partitions
This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new 6eb1c034 [FLINK-35328] Autoscaler supports setting the maximum floating parallelism by the number of Pulsar partitions
6eb1c034 is described below
commit 6eb1c034cdbf678f595b7a5725476716c0299c55
Author: wenbingshen <ol...@gmail.com>
AuthorDate: Tue Jun 11 14:09:29 2024 +0800
[FLINK-35328] Autoscaler supports setting the maximum floating parallelism by the number of Pulsar partitions
---
.../flink/autoscaler/ScalingMetricCollector.java | 33 ++++++++++++++++++----
.../MetricsCollectionAndEvaluationTest.java | 21 +++++++++++++-
2 files changed, 47 insertions(+), 7 deletions(-)
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java
index 276bf561..ed7906c4 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java
@@ -58,12 +58,14 @@ import java.util.HashMap;
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 @@ public abstract class ScalingMetricCollector<KEY, Context extends JobAutoScalerC
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 @@ public abstract class ScalingMetricCollector<KEY, Context extends JobAutoScalerC
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 {}",
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/MetricsCollectionAndEvaluationTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/MetricsCollectionAndEvaluationTest.java
index df2b1049..ade85b6e 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/MetricsCollectionAndEvaluationTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/MetricsCollectionAndEvaluationTest.java
@@ -232,7 +232,7 @@ public class MetricsCollectionAndEvaluationTest {
}
@Test
- public void testKafkaPartitionMaxParallelism() throws Exception {
+ public void testKafkaPulsarPartitionMaxParallelism() throws Exception {
setDefaultMetrics(metricsCollector);
metricsCollector.updateMetrics(context, stateStore);
@@ -261,6 +261,25 @@ public class MetricsCollectionAndEvaluationTest {
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