@Test
    public void testTimeoutMetrics() throws Throwable {
        try (
            LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(3).
                build();
            QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv).
                build()
        ) {
            QuorumController active = controlEnv.activeController();
            Map<Integer, Long> brokerEpochs = registerBrokersAndUnfence(active, 3);
            assertEquals(0L, active.controllerMetrics().timedOutHeartbeats());
            assertEquals(0L, active.controllerMetrics().operationsTimedOut());

            // We pause the controller so that the heartbeat event will definitely be expired
            // rather than processed.
            CountDownLatch latch = pause(active);
            ControllerRequestContext expiredTimeoutContext = new ControllerRequestContext(
                new RequestHeaderData(),
                KafkaPrincipal.ANONYMOUS,
                OptionalLong.of(active.time().nanoseconds()));
            CompletableFuture<BrokerHeartbeatReply> replyFuture =
                active.processBrokerHeartbeat(expiredTimeoutContext,
                    new BrokerHeartbeatRequestData()
                        .setWantFence(false)
                        .setBrokerEpoch(brokerEpochs.get(0))
                        .setBrokerId(0)
                        .setCurrentMetadataOffset(100000));
            latch.countDown(); // Unpause the controller.
            assertEquals(TimeoutException.class,
                assertThrows(ExecutionException.class, replyFuture::get).
                    getCause().getClass());
            assertEquals(1L, active.controllerMetrics().timedOutHeartbeats());
            assertEquals(1L, active.controllerMetrics().operationsTimedOut());

            // Inject a new timed out operation.
            CountDownLatch latch2 = pause(active);
            active.appendControlEventWithDeadline("fakeTimeoutOperation",
                () -> { },
                active.time().nanoseconds());
            latch2.countDown();
            TestUtils.retryOnExceptionWithTimeout(30_000, () -> {
                // The fake timeout increments operationsTimedOut but not timedOutHeartbeats.
                assertEquals(1L, active.controllerMetrics().timedOutHeartbeats());
                assertEquals(2L, active.controllerMetrics().operationsTimedOut());
            });
            for (QuorumController controller : controlEnv.controllers()) {
                // Inactive controllers don't set these metrics.
                if (!controller.isActive()) {
                    assertFalse(controller.controllerMetrics().active());
                    assertEquals(0L, controller.controllerMetrics().timedOutHeartbeats());
                    assertEquals(0L, controller.controllerMetrics().operationsTimedOut());
                }
            }
        }
    }
Test the heartbeat and general operation timeout metrics.
These are incremented on the active controller only.
metadata/src/test/java/org/apache/kafka/controller/QuorumControllerMetricsIntegrationTest.java