@Test
    public void testHeartbeatMetrics() {
        // setup
        coordinatorRequestManager = mock(CoordinatorRequestManager.class);
        membershipManager = mock(ShareMembershipManager.class);
        heartbeatState = mock(ShareHeartbeatRequestManager.HeartbeatState.class);
        time = new MockTime();
        metrics = new Metrics(time);
        heartbeatRequestState = new ShareHeartbeatRequestManager.HeartbeatRequestState(
                new LogContext(),
                time,
                0, // This initial interval should be 0 to ensure heartbeat on the clock
                retryBackoffMs,
                retryBackoffMaxMs,
                0);
        backgroundEventHandler = mock(BackgroundEventHandler.class);
        heartbeatRequestManager = createHeartbeatRequestManager(
                coordinatorRequestManager,
                membershipManager,
                heartbeatState,
                heartbeatRequestState,
                backgroundEventHandler);
        when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(new Node(1, "localhost", 9999)));
        when(membershipManager.state()).thenReturn(MemberState.STABLE);

        assertNotNull(getMetric("heartbeat-response-time-max"));
        assertNotNull(getMetric("heartbeat-rate"));
        assertNotNull(getMetric("heartbeat-total"));
        assertNotNull(getMetric("last-heartbeat-seconds-ago"));

        // test poll
        assertHeartbeat(heartbeatRequestManager, 0);
        time.sleep(heartbeatIntervalMs);
        assertEquals(1.0, getMetric("heartbeat-total").metricValue());
        assertEquals((double) TimeUnit.MILLISECONDS.toSeconds(heartbeatIntervalMs), getMetric("last-heartbeat-seconds-ago").metricValue());

        assertHeartbeat(heartbeatRequestManager, heartbeatIntervalMs);
        assertEquals(0.06d, (double) getMetric("heartbeat-rate").metricValue(), 0.005d);
        assertEquals(2.0, getMetric("heartbeat-total").metricValue());

        // Randomly sleep for some time
        Random rand = new Random();
        int randomSleepS = rand.nextInt(11);
        time.sleep(randomSleepS * 1000);
        assertEquals((double) randomSleepS, getMetric("last-heartbeat-seconds-ago").metricValue());
    }
This is expected to be the case where a member is already leaving the group and the poll
timer expires. The poll timer expiration should not transition the member to STALE, and
the member should continue to send heartbeats while the ongoing leaving operation
completes (send heartbeats while waiting for callbacks before leaving, or send last
heartbeat to leave).
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareHeartbeatRequestManagerTest.java