This is an automated email from the ASF dual-hosted git repository.
bowenli86 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
The following commit(s) were added to refs/heads/main by this push:
new 2b8d8e20 [FLINK-39990] Expose DynamicKafkaSource active split count
(#277)
2b8d8e20 is described below
commit 2b8d8e20439ae73c86d7f509119935b6be2f5f38
Author: lnbest0707 <[email protected]>
AuthorDate: Wed Jun 24 22:46:24 2026 -0700
[FLINK-39990] Expose DynamicKafkaSource active split count (#277)
This change adds a lifetime DynamicKafkaSource.activeSplitCount gauge for
each dynamic source
reader. It reports 0 before assignment and after all active splits are
removed. The count is
derived only from splits assigned to active underlying KafkaSourceReaders,
so pending restored
state and retained removed-cluster checkpoint offsets are excluded. The
cached atomic value keeps
metric reporter threads from traversing mutable reader state.
---
.../docs/connectors/datastream/dynamic-kafka.md | 8 ++-
.../docs/connectors/datastream/dynamic-kafka.md | 8 ++-
.../source/reader/DynamicKafkaSourceReader.java | 29 +++++++++
.../dynamic/source/DynamicKafkaSourceITTest.java | 71 +++++++++-------------
.../reader/DynamicKafkaSourceReaderTest.java | 71 +++++++++++++++++++++-
5 files changed, 143 insertions(+), 44 deletions(-)
diff --git a/docs/content.zh/docs/connectors/datastream/dynamic-kafka.md
b/docs/content.zh/docs/connectors/datastream/dynamic-kafka.md
index 8f4adc6d..a55264b8 100644
--- a/docs/content.zh/docs/connectors/datastream/dynamic-kafka.md
+++ b/docs/content.zh/docs/connectors/datastream/dynamic-kafka.md
@@ -296,7 +296,7 @@ a list of applicable properties.
</thead>
<tbody>
<tr>
- <th rowspan="8">Operator</th>
+ <th rowspan="6">Operator</th>
<td>currentEmitEventTimeLag</td>
<td>n/a</td>
<td>The time span from the record event timestamp to the time the
record is emitted by the source connector¹: <code>currentEmitEventTimeLag =
EmitTime - EventTime.</code></td>
@@ -326,6 +326,12 @@ a list of applicable properties.
<td>The total number of Kafka clusters read by this reader.</td>
<td>Gauge</td>
</tr>
+ <tr>
+ <td>activeSplitCount</td>
+ <td>n/a</td>
+ <td>当前分配给此 source reader 的活跃 split 数。该 gauge 在 reader
的整个生命周期内保持注册；当元数据移除所有本地 split 后会报告 <code>0</code>。为已移除集群保留的 checkpoint offset
不计入其中。</td>
+ <td>Gauge</td>
+ </tr>
</tbody>
</table>
diff --git a/docs/content/docs/connectors/datastream/dynamic-kafka.md
b/docs/content/docs/connectors/datastream/dynamic-kafka.md
index 8df05afb..b3bc2073 100644
--- a/docs/content/docs/connectors/datastream/dynamic-kafka.md
+++ b/docs/content/docs/connectors/datastream/dynamic-kafka.md
@@ -305,7 +305,7 @@ a list of applicable properties.
</thead>
<tbody>
<tr>
- <th rowspan="8">Operator</th>
+ <th rowspan="6">Operator</th>
<td>currentEmitEventTimeLag</td>
<td>n/a</td>
<td>The time span from the record event timestamp to the time the
record is emitted by the source connector¹: <code>currentEmitEventTimeLag =
EmitTime - EventTime.</code></td>
@@ -335,6 +335,12 @@ a list of applicable properties.
<td>The total number of Kafka clusters read by this reader.</td>
<td>Gauge</td>
</tr>
+ <tr>
+ <td>activeSplitCount</td>
+ <td>n/a</td>
+ <td>The number of currently assigned active splits in this source
reader. The gauge is registered for the reader lifetime and reports
<code>0</code> after metadata removes all local splits. Retained
removed-cluster checkpoint offsets are excluded.</td>
+ <td>Gauge</td>
+ </tr>
</tbody>
</table>
diff --git
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java
index f5103392..e2a3a572 100644
---
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java
+++
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java
@@ -69,6 +69,7 @@ import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
/**
@@ -83,10 +84,13 @@ import java.util.stream.Collectors;
@Internal
public class DynamicKafkaSourceReader<T> implements SourceReader<T,
DynamicKafkaSourceSplit> {
private static final Logger logger =
LoggerFactory.getLogger(DynamicKafkaSourceReader.class);
+ static final String ACTIVE_SPLIT_COUNT_METRIC = "activeSplitCount";
+
private final KafkaRecordDeserializationSchema<T> deserializationSchema;
private final Properties properties;
private final MetricGroup dynamicKafkaSourceMetricGroup;
private final Gauge<Integer> kafkaClusterCount;
+ private final AtomicInteger activeSplitCount;
private final SourceReaderContext readerContext;
private final KafkaClusterMetricGroupManager
kafkaClusterMetricGroupManager;
@@ -114,6 +118,7 @@ public class DynamicKafkaSourceReader<T> implements
SourceReader<T, DynamicKafka
this.deserializationSchema = deserializationSchema;
this.properties = properties;
this.kafkaClusterCount = clusterReaderMap::size;
+ this.activeSplitCount = new AtomicInteger();
this.dynamicKafkaSourceMetricGroup =
readerContext
.metricGroup()
@@ -141,6 +146,10 @@ public class DynamicKafkaSourceReader<T> implements
SourceReader<T, DynamicKafka
logger.trace("Starting reader for subtask index={}",
readerContext.getIndexOfSubtask());
// metrics cannot be registered in the enumerator
readerContext.metricGroup().gauge("kafkaClusterCount",
kafkaClusterCount);
+ // Keep this gauge registered for the full source-reader lifetime. In
particular, it must
+ // continue reporting zero after metadata removes all locally assigned
splits, rather than
+ // relying on split-specific metric groups that disappear with removed
clusters.
+ dynamicKafkaSourceMetricGroup.gauge(ACTIVE_SPLIT_COUNT_METRIC,
activeSplitCount::get);
readerContext.sendSourceEventToCoordinator(new
GetMetadataUpdateEvent());
}
@@ -174,6 +183,7 @@ public class DynamicKafkaSourceReader<T> implements
SourceReader<T, DynamicKafka
}
}
+ refreshActiveSplitCount();
return logAndReturnInputStatus(consolidateInputStatus(isMoreAvailable,
isNothingAvailable));
}
@@ -239,6 +249,7 @@ public class DynamicKafkaSourceReader<T> implements
SourceReader<T, DynamicKafka
if (newCluster) {
completeAndResetAvailabilityHelper();
}
+ refreshActiveSplitCount();
}
/**
@@ -343,6 +354,7 @@ public class DynamicKafkaSourceReader<T> implements
SourceReader<T, DynamicKafka
// reset the availability future to also depend on the new sub
readers
completeAndResetAvailabilityHelper();
+ refreshActiveSplitCount();
} else {
// update properties even on no metadata change
clustersProperties.clear();
@@ -479,6 +491,7 @@ public class DynamicKafkaSourceReader<T> implements
SourceReader<T, DynamicKafka
for (KafkaSourceReader<T> subReader : clusterReaderMap.values()) {
subReader.close();
}
+ activeSplitCount.set(0);
kafkaClusterMetricGroupManager.close();
}
@@ -630,6 +643,22 @@ public class DynamicKafkaSourceReader<T> implements
SourceReader<T, DynamicKafka
}
clusterReaderMap.clear();
clustersProperties.clear();
+ // Keep the last published count during a rebuild because metric
reporters may sample the
+ // gauge concurrently. The caller publishes the final count after
replacement readers have
+ // received their splits.
+ }
+
+ /**
+ * Refresh the local active split count from the underlying readers on the
source reader's main
+ * thread. Pending restored splits and retained removed-cluster offsets
are intentionally not
+ * counted until metadata has validated and assigned them to an active
reader.
+ */
+ private void refreshActiveSplitCount() {
+ int currentActiveSplitCount =
+ clusterReaderMap.values().stream()
+
.mapToInt(KafkaSourceReader::getNumberOfCurrentlyAssignedSplits)
+ .sum();
+ activeSplitCount.set(currentActiveSplitCount);
}
private void reactivateRetainedSplits(
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/DynamicKafkaSourceITTest.java
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/DynamicKafkaSourceITTest.java
index 3bb77e7c..eafbf19c 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/DynamicKafkaSourceITTest.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/DynamicKafkaSourceITTest.java
@@ -58,7 +58,6 @@ import
org.apache.flink.connector.testframe.testsuites.SourceTestSuiteBase;
import org.apache.flink.core.execution.CheckpointingMode;
import org.apache.flink.core.execution.JobClient;
import org.apache.flink.core.testutils.CommonTestUtils;
-import org.apache.flink.metrics.MetricGroup;
import
org.apache.flink.runtime.messages.FlinkJobTerminatedWithoutCancellationException;
import org.apache.flink.runtime.testutils.InMemoryReporter;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
@@ -100,7 +99,6 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
-import java.util.Optional;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;
@@ -1120,15 +1118,8 @@ class DynamicKafkaSourceITTest {
.boxed()
.collect(Collectors.toList()));
- // should contain cluster 0 metrics
- assertThat(findMetrics(reporter,
DYNAMIC_KAFKA_SOURCE_METRIC_GROUP))
- .allSatisfy(
- metricName ->
- assertThat(metricName)
- .containsPattern(
- ".*"
- +
DYNAMIC_KAFKA_SOURCE_METRIC_GROUP
- +
"\\.kafkaCluster\\.kafka-cluster-0.*"));
+ // should contain only cluster 0 metrics
+ waitForOnlyKafkaClusterMetrics("kafka-cluster-0");
// setup test data for cluster 1 and stop consuming from
cluster 0
latestValueOffset =
@@ -1148,24 +1139,8 @@ class DynamicKafkaSourceITTest {
results.add(iterator.next());
}
- // cluster 0 is not being consumed from, metrics should not
appear
- assertThat(findMetrics(reporter,
DYNAMIC_KAFKA_SOURCE_METRIC_GROUP))
- .allSatisfy(
- metricName ->
- assertThat(metricName)
- .doesNotContainPattern(
- ".*"
- +
DYNAMIC_KAFKA_SOURCE_METRIC_GROUP
- +
"\\.kafkaCluster\\.kafka-cluster-0.*"));
-
- assertThat(findMetrics(reporter,
DYNAMIC_KAFKA_SOURCE_METRIC_GROUP))
- .allSatisfy(
- metricName ->
- assertThat(metricName)
- .containsPattern(
- ".*"
- +
DYNAMIC_KAFKA_SOURCE_METRIC_GROUP
- +
"\\.kafkaCluster\\.kafka-cluster-1.*"));
+ // cluster 0 is not being consumed from, metrics should
contain only cluster 1
+ waitForOnlyKafkaClusterMetrics("kafka-cluster-1");
}
}
@@ -1235,11 +1210,16 @@ class DynamicKafkaSourceITTest {
kafkaClusterTestEnvMetadataList);
}
- private Set<String> findMetrics(InMemoryReporter inMemoryReporter,
String groupPattern) {
- Optional<MetricGroup> groups =
inMemoryReporter.findGroup(groupPattern);
- assertThat(groups).isPresent();
- return
inMemoryReporter.getMetricsByGroup(groups.get()).keySet().stream()
- .map(metricName ->
groups.get().getMetricIdentifier(metricName))
+ private Set<String> findKafkaClusterMetrics(InMemoryReporter
inMemoryReporter) {
+ // Metrics are registered per source subtask, so aggregate every
matching group.
+ return
inMemoryReporter.findGroups(DYNAMIC_KAFKA_SOURCE_METRIC_GROUP).stream()
+ .flatMap(
+ group ->
+
inMemoryReporter.getMetricsByGroup(group).keySet().stream()
+ .map(
+ metricName ->
+
group.getMetricIdentifier(metricName)))
+ .filter(metricName ->
metricName.contains(".kafkaCluster."))
.collect(Collectors.toSet());
}
@@ -1337,14 +1317,23 @@ class DynamicKafkaSourceITTest {
"Could not observe removed Kafka cluster metrics
disappear");
}
- private boolean hasKafkaClusterMetrics(String kafkaClusterId) {
- Optional<MetricGroup> groups =
reporter.findGroup(DYNAMIC_KAFKA_SOURCE_METRIC_GROUP);
- if (groups.isEmpty()) {
- return false;
- }
+ private void waitForOnlyKafkaClusterMetrics(String kafkaClusterId)
throws Exception {
+ CommonTestUtils.waitUtil(
+ () -> {
+ Set<String> metrics =
findKafkaClusterMetrics(reporter);
+ return !metrics.isEmpty()
+ && metrics.stream()
+ .allMatch(
+ metricName ->
+ metricName.contains(
+
".kafkaCluster." + kafkaClusterId));
+ },
+ Duration.ofSeconds(30),
+ "Could not observe only Kafka cluster metrics for " +
kafkaClusterId);
+ }
- return reporter.getMetricsByGroup(groups.get()).keySet().stream()
- .map(metricName ->
groups.get().getMetricIdentifier(metricName))
+ private boolean hasKafkaClusterMetrics(String kafkaClusterId) {
+ return findKafkaClusterMetrics(reporter).stream()
.anyMatch(metricName ->
metricName.contains(".kafkaCluster." + kafkaClusterId));
}
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReaderTest.java
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReaderTest.java
index c8e95ef3..180c3392 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReaderTest.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReaderTest.java
@@ -26,9 +26,11 @@ import
org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
import org.apache.flink.connector.base.source.reader.splitreader.SplitsChange;
import
org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
+import org.apache.flink.connector.kafka.dynamic.metadata.ClusterMetadata;
import org.apache.flink.connector.kafka.dynamic.metadata.KafkaStream;
import
org.apache.flink.connector.kafka.dynamic.source.DynamicKafkaSourceOptions;
import org.apache.flink.connector.kafka.dynamic.source.MetadataUpdateEvent;
+import
org.apache.flink.connector.kafka.dynamic.source.metrics.KafkaClusterMetricGroup;
import
org.apache.flink.connector.kafka.dynamic.source.split.DynamicKafkaSourceSplit;
import
org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderMetrics;
import org.apache.flink.connector.kafka.source.reader.KafkaSourceReader;
@@ -40,7 +42,10 @@ import
org.apache.flink.connector.testutils.source.reader.SourceReaderTestBase;
import org.apache.flink.connector.testutils.source.reader.TestingReaderContext;
import org.apache.flink.connector.testutils.source.reader.TestingReaderOutput;
import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.flink.metrics.testutils.MetricListener;
+import org.apache.flink.runtime.metrics.groups.InternalSourceReaderMetricGroup;
import
org.apache.flink.streaming.connectors.kafka.DynamicKafkaSourceTestHelper;
import com.google.common.collect.ImmutableList;
@@ -63,6 +68,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
+import java.util.Optional;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
@@ -112,6 +118,65 @@ public class DynamicKafkaSourceReaderTest extends
SourceReaderTestBase<DynamicKa
DynamicKafkaSourceTestHelper.tearDown();
}
+ @Test
+ void testActiveSplitCountMetricTracksMetadataRemoval() throws Exception {
+ MetricListener metricListener = new MetricListener();
+ TestingReaderContext context =
+ new TestingReaderContext(
+ new Configuration(),
+
InternalSourceReaderMetricGroup.mock(metricListener.getMetricGroup()));
+ Properties properties = getRequiredProperties();
+ properties.setProperty(
+
DynamicKafkaSourceOptions.STREAM_METADATA_REMOVED_CLUSTER_RETENTION_MS.key(),
+ "60000");
+
+ try (DynamicKafkaSourceReader<Integer> reader =
+ createReaderWithoutStart(context, properties)) {
+ reader.start();
+ Optional<Gauge<Integer>> activeSplitCountGauge =
+ metricListener.getGauge(
+
KafkaClusterMetricGroup.DYNAMIC_KAFKA_SOURCE_METRIC_GROUP,
+
DynamicKafkaSourceReader.ACTIVE_SPLIT_COUNT_METRIC);
+ assertThat(activeSplitCountGauge).isPresent();
+
assertThat(activeSplitCountGauge.orElseThrow().getValue()).isZero();
+
+
reader.handleSourceEvents(DynamicKafkaSourceTestHelper.getMetadataUpdateEvent(TOPIC));
+ assertThat(activeSplitCountGauge.orElseThrow().getValue())
+ .as("cluster readers without assigned splits remain empty")
+ .isZero();
+ reader.addSplits(
+ getSplits(
+ getNumSplits(),
+ NUM_RECORDS_PER_SPLIT,
+ Boundedness.CONTINUOUS_UNBOUNDED));
+
assertThat(activeSplitCountGauge.orElseThrow().getValue()).isEqualTo(getNumSplits());
+
+ KafkaStream kafkaStream =
DynamicKafkaSourceTestHelper.getKafkaStream(TOPIC);
+ ClusterMetadata cluster1Metadata =
+ kafkaStream.getClusterMetadataMap().get(kafkaClusterId1);
+ kafkaStream.getClusterMetadataMap().remove(kafkaClusterId0);
+ reader.handleSourceEvents(new
MetadataUpdateEvent(Collections.singleton(kafkaStream)));
+ assertThat(activeSplitCountGauge.orElseThrow().getValue())
+ .isEqualTo(NUM_SPLITS_PER_CLUSTER);
+ assertThat(reader.snapshotState(-1))
+ .as("retained removed-cluster state is not an active
split")
+ .hasSize(getNumSplits());
+
+ kafkaStream.getClusterMetadataMap().remove(kafkaClusterId1);
+ reader.handleSourceEvents(new
MetadataUpdateEvent(Collections.singleton(kafkaStream)));
+
assertThat(activeSplitCountGauge.orElseThrow().getValue()).isZero();
+ assertThat(reader.snapshotState(-1))
+ .as("all removed-cluster state is retained but inactive")
+ .hasSize(getNumSplits());
+
+ kafkaStream.getClusterMetadataMap().put(kafkaClusterId1,
cluster1Metadata);
+ reader.handleSourceEvents(new
MetadataUpdateEvent(Collections.singleton(kafkaStream)));
+ assertThat(activeSplitCountGauge.orElseThrow().getValue())
+ .as("reactivated retained splits become active again")
+ .isEqualTo(NUM_SPLITS_PER_CLUSTER);
+ }
+ }
+
@Test
void testHandleSourceEventWithRemovedMetadataAtStartup() throws Exception {
TestingReaderContext context = new TestingReaderContext();
@@ -392,7 +457,11 @@ public class DynamicKafkaSourceReaderTest extends
SourceReaderTestBase<DynamicKa
private DynamicKafkaSourceReader<Integer> createReaderWithoutStart(
TestingReaderContext context) {
- Properties properties = getRequiredProperties();
+ return createReaderWithoutStart(context, getRequiredProperties());
+ }
+
+ private DynamicKafkaSourceReader<Integer> createReaderWithoutStart(
+ TestingReaderContext context, Properties properties) {
return new DynamicKafkaSourceReader<>(
context,
KafkaRecordDeserializationSchema.valueOnly(IntegerDeserializer.class),
