soarez commented on a change in pull request #8935:
URL: https://github.com/apache/kafka/pull/8935#discussion_r454207976
##########
File path:
core/src/test/scala/unit/kafka/controller/ControllerEventManagerTest.scala
##########
@@ -135,6 +135,43 @@ 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()
+ processedEvents.incrementAndGet()
+ }
+ override def preempt(event: ControllerEvent): Unit = {}
+ }
+
+ controllerEventManager = new ControllerEventManager(0, eventProcessor,
+ time, controllerStats.rateAndTimeMetrics, 1)
+ controllerEventManager.start()
+
+ controllerEventManager.put(TopicChange)
+ controllerEventManager.put(TopicChange)
+ latch.countDown()
Review comment:
Why is this `CountdownLatch` necessary? Seems it’s just delaying the
wait condition being true - `processedEvents.get() == 2`
----------------------------------------------------------------
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
