KAFKA-20363 : server-common - Fix flaky testIdleTimeCallback in KafkaEventQueueTest

server-common/src/test/java/org/apache/kafka/queue/KafkaEventQueueTest.java

@@ -440,6 +440,11 @@ public void testIdleTimeCallback() throws Exception {
                     lastIdleTimeMs.set(idleDuration);
                     lastCurrentTimeMs.set(currentTime);
                 })) {
+            // Capture the queue's event handler thread so we can wait for it to be idle.
+            CompletableFuture<Thread> queueThreadFuture = new CompletableFuture<>();
+            queue.append(() -> queueThreadFuture.complete(Thread.currentThread()));
+            Thread queueThread = queueThreadFuture.get();
+
             time.sleep(2);
             assertEquals(0, lastIdleTimeMs.get(), "Last idle time should be 0ms");
 
@@ -451,6 +456,13 @@ public void testIdleTimeCallback() throws Exception {
             }));
             assertEquals("event1-processed", event1.get());
 
+            // Wait for the queue thread to enter cond.await() before advancing MockTime.
+            // This ensures startIdleMs is captured before the time advancement.
+            TestUtils.waitForCondition(
+                    () -> queueThread.getState() == Thread.State.WAITING,
+                    "Queue thread should be waiting"
+            );
+
             long timeBeforeWait = time.milliseconds();
             long waitTime5Ms = 5;
             time.sleep(waitTime5Ms);
@@ -464,6 +476,12 @@ public void testIdleTimeCallback() throws Exception {
             assertEquals(timeBeforeWait + waitTime5Ms, lastCurrentTimeMs.get(), "Current time should be " + (timeBeforeWait + waitTime5Ms) + "ms, was: " + lastCurrentTimeMs.get());
 
             // Test 2: Deferred event
+            // Wait for the queue thread to enter cond.await() before advancing MockTime.
+            TestUtils.waitForCondition(
+                    () -> queueThread.getState() == Thread.State.WAITING,
+                    "Queue thread should be waiting"
+            );
+
             long timeBeforeDeferred = time.milliseconds();
             long waitTime2Ms = 2;
             CompletableFuture<Void> deferredEvent2 = new CompletableFuture<>();