0xffff-zhiyan commented on code in PR #20481:
URL: https://github.com/apache/kafka/pull/20481#discussion_r2407137104
##########
core/src/test/scala/kafka/server/KafkaRequestHandlerTest.scala:
##########
@@ -698,4 +699,72 @@ class KafkaRequestHandlerTest {
// cleanup
brokerTopicStats.close()
}
+
+ @Test
+ def testGlobalSharedThreadCounter(): Unit = {
+ val time = Time.SYSTEM
+ val metricsBroker = new RequestChannelMetrics(java.util.Set.of[ApiKeys])
+ val metricsController = new
RequestChannelMetrics(java.util.Set.of[ApiKeys])
+ val requestChannelBroker = new RequestChannel(10, time, metricsBroker)
+ val requestChannelController = new RequestChannel(10, time,
metricsController)
+ val apiHandler = mock(classOf[ApiRequestHandler])
+
+ // Reset global shared counter for test
+ KafkaRequestHandlerPool.sharedAggregateTotalThreads.set(0)
+
+ // Create broker pool with 4 threads
+ val brokerPool = new KafkaRequestHandlerPool(
+ 0,
+ requestChannelBroker,
+ apiHandler,
+ time,
+ 4,
+ "RequestHandlerAvgIdlePercent",
+ "broker"
+ )
+
+ // Verify global counter is updated
+ assertEquals(4, KafkaRequestHandlerPool.sharedAggregateTotalThreads.get,
"global counter should be 4 after broker pool")
+
+ // Create controller pool with 4 threads
+ val controllerPool = new KafkaRequestHandlerPool(
+ 0,
+ requestChannelController,
+ apiHandler,
+ time,
+ 4,
+ "RequestHandlerAvgIdlePercent",
+ "controller"
+ )
+
+ // Verify global counter is updated to sum of both pools
+ assertEquals(8, KafkaRequestHandlerPool.sharedAggregateTotalThreads.get,
"global counter should be 8 after both pools")
+
+ try {
+ // Get the aggregate meter from broker pool using reflection
+ val aggregateMeterField =
classOf[KafkaRequestHandlerPool].getDeclaredField("aggregateIdleMeter")
+ aggregateMeterField.setAccessible(true)
+ val brokerAggregateMeter =
aggregateMeterField.get(brokerPool).asInstanceOf[Meter]
+
+ // Wait for idle measurements to accumulate
+ val deadline = System.currentTimeMillis() + 8000
+ var value = 0.0
+ while (System.currentTimeMillis() < deadline && value == 0.0) {
+ Thread.sleep(200)
+ value = brokerAggregateMeter.oneMinuteRate()
+ }
+ // Verify that the aggregate meter shows reasonable idle percentage
+ // Since both pools are hitting the same global counter (8 threads), the
rate should be normalized
+ assertTrue(value >= 0.0 && value <= 1.05, s"aggregate idle percent
should be within [0,1], got $value")
Review Comment:
Yes I agree. I think the sleep(200) is the reason why I observed the value
sightly greater than 1(like 1.00xx, only seen once). The test samples
oneMinuteRate() every 200ms, but the algorithm(called EWMA) needs some time to
properly digest new data points. During idle periods, when large idle time
values are suddenly recorded, the algorithm goes through a brief overshoot
phase before stabilizing. A longer sleep interval (1-2 seconds) would likely
eliminate this issue.
--
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]
