You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by mx...@apache.org on 2025/09/19 11:27:21 UTC
(flink-kubernetes-operator) branch main updated: [FLINK-38381] Enumerate Kafka partitions across Kafka clusters (#1030)
This is an automated email from the ASF dual-hosted git repository.
mxm 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 003e594c [FLINK-38381] Enumerate Kafka partitions across Kafka clusters (#1030)
003e594c is described below
commit 003e594c9a52ac67fb64ef67588291fcbc85c99e
Author: Maximilian Michels <mx...@apache.org>
AuthorDate: Fri Sep 19 13:27:15 2025 +0200
[FLINK-38381] Enumerate Kafka partitions across Kafka clusters (#1030)
When using the DynamicKafkaSink, topics can be spread across multiple
clusters. This used to work fine, but a regression has been added which
considers partitions across different clusters to be identical. This limits the
scale out of the source operator.
Here is an example:
```
"1.Source__Kafka_Source_(testTopic).kafkaCluster.my-cluster-1.KafkaSourceReader.topic.testTopic.partition.0.currentOffset",
"1.Source__Kafka_Source_(testTopic).kafkaCluster.my-cluster-2.KafkaSourceReader.topic.testTopic.partition.0.currentOffset"
```
Those would result be treated as one partition, but there are two partitions from separate kafka clusters.
---
.../apache/flink/autoscaler/ScalingMetricCollector.java | 10 ++++++++--
.../autoscaler/MetricsCollectionAndEvaluationTest.java | 14 ++++++++++++++
2 files changed, 22 insertions(+), 2 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 ee3ded9e..310e063b 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
@@ -269,7 +269,7 @@ public abstract class ScalingMetricCollector<KEY, Context extends JobAutoScalerC
try (var restClient = ctx.getRestClusterClient()) {
Pattern partitionRegex =
Pattern.compile(
- "^.*\\.KafkaSourceReader\\.topic\\.(?<kafkaTopic>.+)\\.partition\\.(?<kafkaId>\\d+)\\.currentOffset$"
+ "^.*?(\\.kafkaCluster\\.(?<kafkaCluster>.+))?\\.KafkaSourceReader\\.topic\\.(?<kafkaTopic>.+)\\.partition\\.(?<kafkaId>\\d+)\\.currentOffset$"
+ "|^.*\\.PulsarConsumer\\.(?<pulsarTopic>.+)-partition-(?<pulsarId>\\d+)\\..*\\.numMsgsReceived$");
for (var vertexInfo : topology.getVertexInfos().values()) {
if (vertexInfo.getInputs().isEmpty()) {
@@ -281,12 +281,18 @@ public abstract class ScalingMetricCollector<KEY, Context extends JobAutoScalerC
Matcher matcher = partitionRegex.matcher(v);
if (matcher.matches()) {
String kafkaTopic = matcher.group("kafkaTopic");
+ String kafkaCluster =
+ matcher.group("kafkaCluster");
String kafkaId = matcher.group("kafkaId");
String pulsarTopic =
matcher.group("pulsarTopic");
String pulsarId = matcher.group("pulsarId");
return kafkaTopic != null
- ? kafkaTopic + "-" + kafkaId
+ ? kafkaCluster
+ + "-"
+ + kafkaTopic
+ + "-"
+ + kafkaId
: pulsarTopic + "-" + pulsarId;
}
return null;
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 1176075f..5bca303f 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
@@ -258,6 +258,20 @@ public class MetricsCollectionAndEvaluationTest {
collectedMetrics = metricsCollector.updateMetrics(context, stateStore);
assertEquals(5, collectedMetrics.getJobTopology().get(source1).getNumSourcePartitions());
+ metricsCollector.setMetricNames(
+ Map.of(
+ source1,
+ List.of(
+ "1.Source__Kafka_Source_(testTopic).kafkaCluster.my-cluster-1.KafkaSourceReader.topic.testTopic.partition.0.currentOffset",
+ "1.Source__Kafka_Source_(testTopic).kafkaCluster.my-cluster-1.KafkaSourceReader.topic.anotherTopic.partition.0.currentOffset",
+ "1.Source__Kafka_Source_(testTopic).kafkaCluster.my-cluster-2.KafkaSourceReader.topic.testTopic.partition.0.currentOffset",
+ "1.Source__Kafka_Source_(testTopic).kafkaCluster.my-cluster-2.KafkaSourceReader.topic.testTopic.partition.1.currentOffset",
+ "1.Source__Kafka_Source_(testTopic).kafkaCluster.my-cluster-2.KafkaSourceReader.topic.testTopic.partition.2.currentOffset",
+ "1.Source__Kafka_Source_(testTopic).kafkaCluster.my-cluster-2.KafkaSourceReader.topic.testTopic.partition.3.currentOffset")));
+
+ collectedMetrics = metricsCollector.updateMetrics(context, stateStore);
+ assertEquals(6, collectedMetrics.getJobTopology().get(source1).getNumSourcePartitions());
+
metricsCollector.setMetricNames(
Map.of(
source2,