You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "lnbest0707 (via GitHub)" <gi...@apache.org> on 2026/06/24 18:55:34 UTC
[PR] [FLINK-39990] Expose DynamicKafkaSource active split count [flink-connector-kafka]
lnbest0707 opened a new pull request, #277:
URL: https://github.com/apache/flink-connector-kafka/pull/277
## What is the purpose of the change
`DynamicKafkaSource` currently exposes cluster- and split-scoped metrics whose metric groups can
disappear when dynamic metadata removes clusters or partitions. Consumers such as the Flink
Kubernetes Operator autoscaler need a stable per-reader signal for the current active split
assignment.
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 `KafkaSourceReader`s, 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.
## Brief change log
- register `DynamicKafkaSource.activeSplitCount` at the stable dynamic reader metric root
- refresh the count after split assignment, metadata rebuild/removal, polling, and reader close
- add coverage for empty readers, assignment, partial/all metadata removal, retained-state
exclusion, and reactivation
- keep the existing cluster-metric lifecycle test scoped to cluster metrics
- document the gauge in the English and Chinese Dynamic Kafka source docs
## Verifying this change
```bash
DOCKER_HOST=unix://${HOME}/.colima/default/docker.sock \
TESTCONTAINERS_DOCKER_SOCKET_OVERRIDE=/var/run/docker.sock \
TESTCONTAINERS_HOST_OVERRIDE=127.0.0.1 \
mvn -pl flink-connector-kafka \
-Dtest=DynamicKafkaSourceReaderTest \
-Drat.skip=true clean test
DOCKER_HOST=unix://${HOME}/.colima/default/docker.sock \
TESTCONTAINERS_DOCKER_SOCKET_OVERRIDE=/var/run/docker.sock \
TESTCONTAINERS_HOST_OVERRIDE=127.0.0.1 \
mvn -pl flink-connector-kafka \
-Dtest='DynamicKafkaSourceITTest$DynamicKafkaSourceSpecificTests#testMetricsLifecycleManagement' \
-Drat.skip=true test
DOCKER_HOST=unix://${HOME}/.colima/default/docker.sock \
TESTCONTAINERS_DOCKER_SOCKET_OVERRIDE=/var/run/docker.sock \
TESTCONTAINERS_HOST_OVERRIDE=127.0.0.1 \
mvn -pl flink-connector-kafka -Drat.skip=true clean test
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39990] Expose DynamicKafkaSource active split count [flink-connector-kafka]
Posted by "lnbest0707 (via GitHub)" <gi...@apache.org>.
lnbest0707 commented on code in PR #277:
URL: https://github.com/apache/flink-connector-kafka/pull/277#discussion_r3470701849
##########
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>
Review Comment:
The old number is not correct, it had 5 before, and with the new one, 6 now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39990] Expose DynamicKafkaSource active split count [flink-connector-kafka]
Posted by "bowenli86 (via GitHub)" <gi...@apache.org>.
bowenli86 merged PR #277:
URL: https://github.com/apache/flink-connector-kafka/pull/277
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39990] Expose DynamicKafkaSource active split count [flink-connector-kafka]
Posted by "bowenli-oai (via GitHub)" <gi...@apache.org>.
bowenli-oai commented on code in PR #277:
URL: https://github.com/apache/flink-connector-kafka/pull/277#discussion_r3470452697
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/DynamicKafkaSourceITTest.java:
##########
@@ -1243,6 +1246,12 @@ private Set<String> findMetrics(InMemoryReporter inMemoryReporter, String groupP
.collect(Collectors.toSet());
}
+ private Set<String> findKafkaClusterMetrics(InMemoryReporter inMemoryReporter) {
+ return findMetrics(inMemoryReporter, DYNAMIC_KAFKA_SOURCE_METRIC_GROUP).stream()
Review Comment:
Could we avoid selecting a single arbitrary `DynamicKafkaSource` group here? This PR registers `activeSplitCount` directly on the parent group, so `findGroup(DYNAMIC_KAFKA_SOURCE_METRIC_GROUP)` can return that parent; `getMetricsByGroup()` then contains only the root gauge, and the `.kafkaCluster.` filter yields an empty set. That is consistent with the current 2.2.1/JDK 21 failure at line 1125. Please aggregate `findGroups(...)`/all matching groups or explicitly select kafka-cluster child groups. The same single-group pattern in `hasKafkaClusterMetrics()` can also make disappearance checks pass prematurely.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java:
##########
@@ -630,6 +643,20 @@ private void closeAllReadersAndClearState() {
}
clusterReaderMap.clear();
clustersProperties.clear();
+ activeSplitCount.set(0);
Review Comment:
Can we avoid publishing `0` from the intermediate clear step? A normal metadata rebuild can preserve assigned splits, but metric reporters read this `AtomicInteger` concurrently and can sample `0` after this line before `refreshActiveSplitCount()` publishes the rebuilt count at line 357. For an autoscaler-facing stable signal, that transient is indistinguishable from “all local splits removed.” Since this method’s only caller refreshes after rebuilding, it seems safer to publish only the final count and keep the explicit zero in `close()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39990] Expose DynamicKafkaSource active split count [flink-connector-kafka]
Posted by "bowenli86 (via GitHub)" <gi...@apache.org>.
bowenli86 commented on PR #277:
URL: https://github.com/apache/flink-connector-kafka/pull/277#issuecomment-4794769503
lgtm, merging
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39990] Expose DynamicKafkaSource active split count [flink-connector-kafka]
Posted by "bowenli-oai (via GitHub)" <gi...@apache.org>.
bowenli-oai commented on code in PR #277:
URL: https://github.com/apache/flink-connector-kafka/pull/277#discussion_r3470459510
##########
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>
Review Comment:
what does it do?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org