jsancio commented on code in PR #20481:
URL: https://github.com/apache/kafka/pull/20481#discussion_r2445198099
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -112,7 +114,10 @@ class KafkaRequestHandler(
val req = requestChannel.receiveRequest(300)
val endTime = time.nanoseconds
val idleTime = endTime - startSelectTime
- aggregateIdleMeter.mark(idleTime / totalHandlerThreads.get)
+ // Per-pool idle ratio uses the pool's own thread count as denominator
+ perPoolIdleMeter.mark(idleTime / poolHandlerThreads.get)
+ // Aggregate idle ratio uses the total threads across all pools as
denominator
+ aggregateIdleMeter.mark(idleTime / aggregateThreads.get)
Review Comment:
> passing aggregateThreads through the constructor would require manually
ensuring all pools receive the same reference, while the singleton pattern
guarantees this by design.
You can make sure this is the case by construction. Static dependencies on
object are inflexible and difficult to test.
> aggregateIdleMeter is an independent reporter in each pool instance that
happens to report to the same metric name. So they have different purposes
I don't think this is true. The yammer metrics registry returns the same
meter when you create two meters with the same name.
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -192,19 +197,26 @@ class KafkaRequestHandler(
}
+object KafkaRequestHandlerPool {
+ val aggregateThreads = new AtomicInteger(0)
+ val requestHandlerAvgIdleMetricName = "RequestHandlerAvgIdlePercent"
+}
+
class KafkaRequestHandlerPool(
val brokerId: Int,
val requestChannel: RequestChannel,
val apis: ApiRequestHandler,
time: Time,
numThreads: Int,
- requestHandlerAvgIdleMetricName: String,
nodeName: String = "broker"
) extends Logging {
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 = nodeName.capitalize +
requestHandlerAvgIdleMetricName
Review Comment:
This name is part of Kafka's public API (metric). It would be nice to make
this explicit and hard to change by developers.
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -214,7 +226,18 @@ class KafkaRequestHandlerPool(
}
def createHandler(id: Int): Unit = synchronized {
- runnables += new KafkaRequestHandler(id, brokerId, aggregateIdleMeter,
threadPoolSize, requestChannel, apis, time, nodeName)
+ runnables += new KafkaRequestHandler(
+ id,
+ brokerId,
+ aggregateIdleMeter,
+ perPoolIdleMeter,
+ threadPoolSize,
+ requestChannel,
+ apis,
+ time,
+ nodeName,
+ )
+ aggregateThreads.getAndIncrement()
Review Comment:
Okay. Let's make sure that createHandler and deleteHandler are private
methods.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
