@Override
    public void enqueueLast(CoordinatorEvent event) throws RejectedExecutionException {
        accumulator.addLast(event);
    }
@Test
    public void testMetrics() throws Exception {
        CoordinatorRuntimeMetrics mockRuntimeMetrics = mock(CoordinatorRuntimeMetrics.class);
        Time mockTime = new MockTime();
        AtomicInteger numEventsExecuted = new AtomicInteger(0);

        // Special event which blocks until the latch is released.
        FutureEvent<Integer> blockingEvent = new FutureEvent<>(
            new TopicPartition("foo", 0), () -> {
                mockTime.sleep(4000L);
                return numEventsExecuted.incrementAndGet();
            },
            true,
            mockTime.milliseconds()
        );

        try (MultiThreadedEventProcessor eventProcessor = new MultiThreadedEventProcessor(
            new LogContext(),
            "event-processor-",
            1, // Use a single thread to block event in the processor.
            mockTime,
            mockRuntimeMetrics,
            new DelayEventAccumulator(mockTime, 500L)
        )) {
            // Enqueue the blocking event.
            eventProcessor.enqueueLast(blockingEvent);

            // Ensure that the blocking event is executed.
            waitForCondition(() -> numEventsExecuted.get() > 0,
                "Blocking event not executed.");

            // Enqueue the other event.
            FutureEvent<Integer> otherEvent = new FutureEvent<>(
                new TopicPartition("foo", 0), () -> {
                mockTime.sleep(5000L);
                return numEventsExecuted.incrementAndGet();
            },
                false,
                mockTime.milliseconds()
            );

            eventProcessor.enqueueLast(otherEvent);

            // Pass the time.
            mockTime.sleep(3000L);

            // Events should not be completed.
            assertFalse(otherEvent.future.isDone());

            // Release the blocking event to unblock the thread.
            blockingEvent.release();

            // The blocking event should be completed.
            blockingEvent.future.get(DEFAULT_MAX_WAIT_MS, TimeUnit.SECONDS);
            assertTrue(blockingEvent.future.isDone());
            assertFalse(blockingEvent.future.isCompletedExceptionally());

            // The other event should also be completed.
            otherEvent.future.get(DEFAULT_MAX_WAIT_MS, TimeUnit.SECONDS);
            assertTrue(otherEvent.future.isDone());
            assertFalse(otherEvent.future.isCompletedExceptionally());
            assertEquals(2, numEventsExecuted.get());

            // e1 poll time = 500
            // e1 processing time = 4000
            // e2 enqueue time = 3000
            // e2 poll time = 500
            // e2 processing time = 5000

            // e1 poll time
            verify(mockRuntimeMetrics, times(1)).recordEventQueueTime(500L);
            // e1 processing time + e2 enqueue time
            verify(mockRuntimeMetrics, times(1)).recordEventQueueProcessingTime(7000L);

            // Second event (e2)

            // e1, e2 poll time
            verify(mockRuntimeMetrics, times(2)).recordThreadIdleTime(500L);
            // event queue time = e2 enqueue time + e2 poll time
            verify(mockRuntimeMetrics, times(1)).recordEventQueueTime(3500L);
        }
    }