@Test
    public void testStaleMemberDoesNotSendHeartbeatAndAllowsTransitionToJoiningToRecover() {
        ShareMembershipManager membershipManager = createMemberInStableState();
        doNothing().when(subscriptionState).assignFromSubscribed(any());
        membershipManager.transitionToSendingLeaveGroup(true);
        membershipManager.onHeartbeatRequestSent();
        assertEquals(MemberState.STALE, membershipManager.state());
        assertTrue(membershipManager.shouldSkipHeartbeat(), "Stale member should not send " +
                "heartbeats");
        // Check that a transition to joining is allowed, which is what is expected to happen
        // when the poll timer is reset.
        assertDoesNotThrow(membershipManager::maybeRejoinStaleMember);
    }
Member that intentionally leaves the group (via unsubscribe) should release assignment,
reset epoch to -1, keep member ID, and transition to {@link MemberState#LEAVING} to send out a
heartbeat with the leave epoch. Once the heartbeat request is sent out, the member should
transition to {@link MemberState#UNSUBSCRIBED}
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareMembershipManagerTest.java