private void testLeaveGroupReleasesAssignmentAndResetsEpochToSendLeaveGroup(ShareMembershipManager membershipManager) {
        mockLeaveGroup();

        CompletableFuture<Void> leaveResult = membershipManager.leaveGroup();

        assertEquals(MemberState.LEAVING, membershipManager.state());
        assertFalse(leaveResult.isDone(), "Leave group result should not complete until the " +
                "heartbeat request to leave is sent out.");

        membershipManager.onHeartbeatRequestSent();

        assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());
        assertTrue(leaveResult.isDone());
        assertFalse(leaveResult.isCompletedExceptionally());
        assertEquals(MEMBER_ID, membershipManager.memberId());
        assertEquals(-1, membershipManager.memberEpoch());
        assertTrue(membershipManager.currentAssignment().isEmpty());
        verify(subscriptionState).assignFromSubscribed(Collections.emptySet());
    }
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