root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

KAFKA-19406 Remove BrokerTopicStats#removeOldFollowerMetrics (#19962) 

BTW: whether we should rename
`ReplicaManager#updateLeaderAndFollowerMetrics`

Reviewers: Ken Huang <s7133700@gmail.com>, PoAn Yang
 <payang@apache.org>, TengYao Chi <kitingiao@gmail.com>, Lan Ding
 <isDing_L@163.com>, Chia-Ping Tsai <chia7712@gmail.com>
This commit is contained in:
Xuan-Zhang Gong 2025-06-19 17:57:22 +08:00 committed by GitHub
parent 297bc40d96
commit 79d2c3c62a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 1 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
core/src/main/scala/kafka/server/ReplicaManager.scala
View File

@ -2201,9 +2201,7 @@ class ReplicaManager(val config: KafkaConfig,
private def updateLeaderAndFollowerMetrics(newFollowerTopics: Set[String]): Unit = { private def updateLeaderAndFollowerMetrics(newFollowerTopics: Set[String]): Unit = {
val leaderTopicSet = leaderPartitionsIterator.map(_.topic).toSet val leaderTopicSet = leaderPartitionsIterator.map(_.topic).toSet
newFollowerTopics.diff(leaderTopicSet).foreach(brokerTopicStats.removeOldLeaderMetrics) newFollowerTopics.diff(leaderTopicSet).foreach(brokerTopicStats.removeOldLeaderMetrics)
// Currently, there are no follower metrics that need to be updated.
// remove metrics for brokers which are not followers of a topic
leaderTopicSet.diff(newFollowerTopics).foreach(brokerTopicStats.removeOldFollowerMetrics)
} }
protected[server] def maybeAddLogDirFetchers(partitions: Set[Partition], protected[server] def maybeAddLogDirFetchers(partitions: Set[Partition],

11
storage/src/main/java/org/apache/kafka/storage/log/metrics/BrokerTopicStats.java
View File

@ -76,8 +76,6 @@ public class BrokerTopicStats implements AutoCloseable {
topicMetrics.closeMetric(BrokerTopicMetrics.FAILED_PRODUCE_REQUESTS_PER_SEC); topicMetrics.closeMetric(BrokerTopicMetrics.FAILED_PRODUCE_REQUESTS_PER_SEC);
topicMetrics.closeMetric(BrokerTopicMetrics.TOTAL_PRODUCE_REQUESTS_PER_SEC); topicMetrics.closeMetric(BrokerTopicMetrics.TOTAL_PRODUCE_REQUESTS_PER_SEC);
topicMetrics.closeMetric(BrokerTopicMetrics.PRODUCE_MESSAGE_CONVERSIONS_PER_SEC); topicMetrics.closeMetric(BrokerTopicMetrics.PRODUCE_MESSAGE_CONVERSIONS_PER_SEC);
topicMetrics.closeMetric(BrokerTopicMetrics.REPLICATION_BYTES_OUT_PER_SEC);
topicMetrics.closeMetric(BrokerTopicMetrics.REASSIGNMENT_BYTES_OUT_PER_SEC);
topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_COPY_BYTES_PER_SEC_METRIC.getName()); topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_COPY_BYTES_PER_SEC_METRIC.getName());
topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_FETCH_BYTES_PER_SEC_METRIC.getName()); topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_FETCH_BYTES_PER_SEC_METRIC.getName());
topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_FETCH_REQUESTS_PER_SEC_METRIC.getName()); topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_FETCH_REQUESTS_PER_SEC_METRIC.getName());
@ -99,15 +97,6 @@ public class BrokerTopicStats implements AutoCloseable {
} }
} }
// This method only removes metrics only used for follower
public void removeOldFollowerMetrics(String topic) {
BrokerTopicMetrics topicMetrics = topicStats(topic);
if (topicMetrics != null) {
topicMetrics.closeMetric(BrokerTopicMetrics.REPLICATION_BYTES_IN_PER_SEC);
topicMetrics.closeMetric(BrokerTopicMetrics.REASSIGNMENT_BYTES_IN_PER_SEC);
}
}
public void removeMetrics(String topic) { public void removeMetrics(String topic) {
BrokerTopicMetrics metrics = stats.remove(topic); BrokerTopicMetrics metrics = stats.remove(topic);
if (metrics != null) { if (metrics != null) {