You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "kevin-wu24 (via GitHub)" <gi...@apache.org> on 2025/09/08 16:29:27 UTC
Re: [PR] KAFKA-19606: Fix anomaly of JMX metrics RequestHandlerAvgIdlePercent in kraft combined mode [kafka]
kevin-wu24 commented on code in PR #20481:
URL: https://github.com/apache/kafka/pull/20481#discussion_r2330729812
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -192,6 +197,10 @@ class KafkaRequestHandler(
}
+object KafkaRequestHandlerPool {
+ val sharedAggregateTotalThreads = new AtomicInteger(0)
Review Comment:
`sharedAggregateTotalThreads` is redundant. We can just name this `totalThreads` or `aggregateThreads`.
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -204,17 +213,32 @@ class KafkaRequestHandlerPool(
private val metricsGroup = new KafkaMetricsGroup(this.getClass)
val threadPoolSize: AtomicInteger = new AtomicInteger(numThreads)
- /* a meter to track the average free capacity of the request handlers */
+ /* Per-pool idle meter (broker-only or controller-only) */
+ private val perPoolIdleMeterName = if (nodeName == "controller") "ControllerRequestHandlerAvgIdlePercent" else "BrokerRequestHandlerAvgIdlePercent"
+ private val perPoolIdleMeter = metricsGroup.newMeter(perPoolIdleMeterName, "percent", TimeUnit.NANOSECONDS)
+ /* Aggregate meter to track the average free capacity of the request handlers */
private val aggregateIdleMeter = metricsGroup.newMeter(requestHandlerAvgIdleMetricName, "percent", TimeUnit.NANOSECONDS)
this.logIdent = s"[data-plane Kafka Request Handler on ${nodeName.capitalize} $brokerId] "
val runnables = new mutable.ArrayBuffer[KafkaRequestHandler](numThreads)
+ // when using shared aggregate counter, register this pool's threads
+ sharedAggregateTotalThreads.addAndGet(numThreads)
Review Comment:
Lets move this into the `synchronized` method `createHandler` and call `incrementAndGet` when each thread is created.
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -93,7 +94,8 @@ class KafkaRequestHandler(
val requestChannel: RequestChannel,
apis: ApiRequestHandler,
time: Time,
- nodeName: String = "broker"
+ nodeName: String = "broker",
+ val perPoolIdleMeter: Meter,
Review Comment:
Can we group this with `aggregateIdleMeter` in the class header?
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -204,17 +213,32 @@ class KafkaRequestHandlerPool(
private val metricsGroup = new KafkaMetricsGroup(this.getClass)
val threadPoolSize: AtomicInteger = new AtomicInteger(numThreads)
- /* a meter to track the average free capacity of the request handlers */
+ /* Per-pool idle meter (broker-only or controller-only) */
+ private val perPoolIdleMeterName = if (nodeName == "controller") "ControllerRequestHandlerAvgIdlePercent" else "BrokerRequestHandlerAvgIdlePercent"
+ private val perPoolIdleMeter = metricsGroup.newMeter(perPoolIdleMeterName, "percent", TimeUnit.NANOSECONDS)
+ /* Aggregate meter to track the average free capacity of the request handlers */
Review Comment:
We can also use the constant `RequestThreadIdleMeter.dataPlaneRequestHandlerAvgIdleMetricName`.
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -204,17 +213,32 @@ class KafkaRequestHandlerPool(
private val metricsGroup = new KafkaMetricsGroup(this.getClass)
val threadPoolSize: AtomicInteger = new AtomicInteger(numThreads)
- /* a meter to track the average free capacity of the request handlers */
+ /* Per-pool idle meter (broker-only or controller-only) */
+ private val perPoolIdleMeterName = if (nodeName == "controller") "ControllerRequestHandlerAvgIdlePercent" else "BrokerRequestHandlerAvgIdlePercent"
+ private val perPoolIdleMeter = metricsGroup.newMeter(perPoolIdleMeterName, "percent", TimeUnit.NANOSECONDS)
+ /* Aggregate meter to track the average free capacity of the request handlers */
Review Comment:
If you look at `ControllerServer.scala`, we are not setting `nodeName` when constructing the `KafkaRequestHandlerPool`, so this name is always `Broker...`. We should just prefix `RequestHandlerAvgIdlePercent` with whatever is passed in by `nodeName`
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -192,6 +197,10 @@ class KafkaRequestHandler(
}
+object KafkaRequestHandlerPool {
+ val sharedAggregateTotalThreads = new AtomicInteger(0)
Review Comment:
To me it makes more sense to have this in the `RequestThreadIdleMeter.scala` object file, since that is what is managing the idle metric for all threads. Maybe we should rename the file to `AggregateRequestThreadIdleMeter` to be more explicit.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org