hudeqi commented on code in PR #13929:
URL: https://github.com/apache/kafka/pull/13929#discussion_r1331294105
##########
core/src/main/scala/kafka/server/AbstractFetcherManager.scala:
##########
@@ -39,27 +43,37 @@ abstract class AbstractFetcherManager[T <:
AbstractFetcherThread](val name: Stri
val failedPartitions = new FailedPartitions
this.logIdent = "[" + name + "] "
- private val tags = Map("clientId" -> clientId).asJava
+ newMetrics()
- metricsGroup.newGauge("MaxLag", () => {
- // current max lag across all fetchers/topics/partitions
- fetcherThreadMap.values.foldLeft(0L) { (curMaxLagAll, fetcherThread) =>
- val maxLagThread =
fetcherThread.fetcherLagStats.stats.values.foldLeft(0L)((curMaxLagThread,
lagMetrics) =>
- math.max(curMaxLagThread, lagMetrics.lag))
- math.max(curMaxLagAll, maxLagThread)
- }
- }, tags)
+ private[server] def newMetrics(): Unit = {
+ val tags = Map("clientId" -> clientId).asJava
+ metricsGroup.newGauge(MaxLagMetricName, () => {
Review Comment:
Hi, @nizhikov , thank you for taking the time to review. The significance of
`metricNamesToTags` is that it can be used in the unit test to check whether
all the metrics of the corresponding module are removed when closed, to prevent
someone from adding new metrics but forgetting remove, with this collection,
you only need to add the new metric to this collection.
Removing the leaking metrics is a long and potentially fixed process, as can
be seen from the original
[reviews](https://github.com/apache/kafka/pull/13473#issuecomment-1542365763)
reached with @divijvaidya.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
