tombentley commented on code in PR #12347:
URL: https://github.com/apache/kafka/pull/12347#discussion_r916044426
##########
core/src/main/scala/kafka/log/LogManager.scala:
##########
@@ -307,6 +313,22 @@ class LogManager(logDirs: Seq[File],
log
}
+ import java.util.concurrent.ThreadFactory
+
+ // factory class for naming the log recovery threads used in metrics
+ class LogRecoveryThreadFactory(val baseName: String) extends ThreadFactory {
+ val threadsNum = new AtomicInteger(0)
Review Comment:
Singlular `threadNum`?
##########
core/src/main/scala/kafka/log/LogManager.scala:
##########
@@ -410,12 +437,34 @@ class LogManager(logDirs: Seq[File],
error(s"There was an error in one of the threads during logs loading:
${e.getCause}")
throw e.getCause
} finally {
+ removeLogRecoveryMetrics()
threadPools.foreach(_.shutdown())
}
info(s"Loaded $numTotalLogs logs in ${time.hiResClockMs() - startMs}ms.")
}
+ private[log] def addLogRecoveryMetrics(): Unit = {
+ for (dir <- logDirs) {
+ newGauge("remainingLogsToRecover", () =>
numRemainingLogs.get(dir.getAbsolutePath),
+ Map("dir" -> dir.getAbsolutePath))
+ for (i <- 0 until numRecoveryThreadsPerDataDir) {
+ val threadName = s"log-recovery-${dir.getAbsolutePath}-$i"
+ newGauge("remainingSegmentsToRecover", () =>
numRemainingSegments.get(threadName),
+ Map("dir" -> dir.getAbsolutePath, "threadNum" -> i.toString))
+ }
+ }
+ }
+
+ private[log] def removeLogRecoveryMetrics(): Unit = {
+ for (dir <- logDirs) {
+ removeMetric("remainingLogsToRecover", Map("dir" -> dir.getAbsolutePath))
+ for (i <- 0 until numRecoveryThreadsPerDataDir) {
Review Comment:
`numRecoveryThreadsPerDataDir` can be changed, so what happens if it changes
after the metrics are added and before they're removed?
##########
core/src/main/scala/kafka/log/LogManager.scala:
##########
@@ -410,12 +437,34 @@ class LogManager(logDirs: Seq[File],
error(s"There was an error in one of the threads during logs loading:
${e.getCause}")
throw e.getCause
} finally {
+ removeLogRecoveryMetrics()
threadPools.foreach(_.shutdown())
}
info(s"Loaded $numTotalLogs logs in ${time.hiResClockMs() - startMs}ms.")
}
+ private[log] def addLogRecoveryMetrics(): Unit = {
+ for (dir <- logDirs) {
+ newGauge("remainingLogsToRecover", () =>
numRemainingLogs.get(dir.getAbsolutePath),
+ Map("dir" -> dir.getAbsolutePath))
+ for (i <- 0 until numRecoveryThreadsPerDataDir) {
+ val threadName = s"log-recovery-${dir.getAbsolutePath}-$i"
Review Comment:
Can we encapsulate this in a method, since it's duplicating the logic in the
ThreadFactory?
--
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
