jeffkbkim commented on a change in pull request #9050: URL: https://github.com/apache/kafka/pull/9050#discussion_r459647273
########## File path: core/src/test/scala/unit/kafka/controller/ControllerIntegrationTest.scala ########## @@ -598,6 +603,86 @@ class ControllerIntegrationTest extends ZooKeeperTestHarness { }, "Broker fail to initialize after restart") } + @Test + def testPreemptionOnControllerShutdown(): Unit = { + servers = makeServers(1, enableControlledShutdown = false) + val controller = getController().kafkaController + val count = new AtomicInteger(2) + val latch = new CountDownLatch(1) + val spyThread = spy(controller.eventManager.thread) + controller.eventManager.setControllerEventThread(spyThread) + val processedEvent = new MockEvent(ControllerState.TopicChange) { + override def process(): Unit = latch.await() + override def preempt(): Unit = {} + } + val preemptedEvent = new MockEvent(ControllerState.TopicChange) { + override def process(): Unit = {} + override def preempt(): Unit = count.decrementAndGet() + } + + controller.eventManager.put(processedEvent) + controller.eventManager.put(preemptedEvent) + controller.eventManager.put(preemptedEvent) + + doAnswer((_: InvocationOnMock) => { + latch.countDown() + }).doCallRealMethod().when(spyThread).awaitShutdown() Review comment: @stanislavkozlovski thanks for the comment. i've tried this approach before and the test passes but sometimes output: ``` [2020-07-23 11:12:43,316] ERROR [RequestSendThread controllerId=0] Controller 0 fails to send a request to broker localhost:51542 (id: 0 rack: null) (kafka.controller.RequestSendThread:76) java.lang.InterruptedException ``` i also see a bit of flakiness with this approach as we cannot exactly time when `latch.countDown()` is called. ---------------------------------------------------------------- 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