jsancio commented on a change in pull request #8935:
URL: https://github.com/apache/kafka/pull/8935#discussion_r447303785
##########
File path: core/src/main/scala/kafka/controller/ControllerEventManager.scala
##########
@@ -115,7 +117,17 @@ class ControllerEventManager(controllerId: Int,
logIdent = s"[ControllerEventThread controllerId=$controllerId] "
override def doWork(): Unit = {
- val dequeued = queue.take()
+ val count = eventQueueTimeHist.count()
+ var dequeued: QueuedEvent = null
+ if (count != 0) {
+ dequeued = queue.poll(eventQueueTimeMetricTimeoutMs,
TimeUnit.MILLISECONDS)
+ if (dequeued == null) {
+ eventQueueTimeHist.clear()
+ return
+ }
+ } else {
+ dequeued = queue.take()
+ }
Review comment:
In Scala you can do:
```scala
val dequeued = {
val count = eventQueueTimeHist.count()
if (count != 0) {
val event = queue.poll(..., ...)
if (event == null) {
eventQueueTimeHist.clear()
queue.take()
} else {
event
}
} else {
queue.take()
}
}
```
We can probably move this code to a private method.
##########
File path:
core/src/test/scala/unit/kafka/controller/ControllerEventManagerTest.scala
##########
@@ -135,6 +135,44 @@ class ControllerEventManagerTest {
assertEquals(500, queueTimeHistogram.max, 0.01)
}
+ @Test
+ def testEventQueueTimeResetOnTimeout(): Unit = {
+ val metricName =
"kafka.controller:type=ControllerEventManager,name=EventQueueTimeMs"
+ val controllerStats = new ControllerStats
+ val time = new MockTime()
+ val latch = new CountDownLatch(1)
+ val processedEvents = new AtomicInteger()
+
+ val eventProcessor = new ControllerEventProcessor {
+ override def process(event: ControllerEvent): Unit = {
+ latch.await()
+ time.sleep(500)
Review comment:
I don't think we need this for this test. The test above has this
because they want to check that the `max` is `500`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
