@Test
    public void testHeartbeatThreadClose() throws Exception {
        try (ConsumerCoordinator coordinator = prepareCoordinatorForCloseTest(true, true, groupInstanceId, true)) {
            coordinator.ensureActiveGroup();
            time.sleep(heartbeatIntervalMs + 100);
            Thread.yield(); // Give heartbeat thread a chance to attempt heartbeat
            closeVerifyTimeout(coordinator, Long.MAX_VALUE, requestTimeoutMs, requestTimeoutMs);
            Thread[] threads = new Thread[Thread.activeCount()];
            int threadCount = Thread.enumerate(threads);
            for (int i = 0; i < threadCount; i++) {
                assertFalse(threads[i].getName().contains(groupId), "Heartbeat thread active after close");
            }
        }
    }
Verifies that subscription change updates SubscriptionState correctly even after JoinGroup failures
that don't re-invoke onJoinPrepare.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java