@Test
    public void testLeaveGroupWhenMemberAlreadyLeaving() {
        ShareMembershipManager membershipManager = createMemberInStableState();

        // First leave attempt. Should stay LEAVING until the heartbeat is sent out.
        mockLeaveGroup();
        CompletableFuture<Void> leaveResult1 = membershipManager.leaveGroup();
        assertEquals(MemberState.LEAVING, membershipManager.state());
        verify(subscriptionState).assignFromSubscribed(Collections.emptySet());
        clearInvocations(subscriptionState);

        // Second leave attempt while the first one has not completed yet.
        mockLeaveGroup();
        CompletableFuture<Void> leaveResult2 = membershipManager.leaveGroup();
        assertFalse(leaveResult2.isDone());

        // Complete first leave group operation. Should also complete the second leave group.
        membershipManager.onHeartbeatRequestSent();
        assertTrue(leaveResult1.isDone());
        assertFalse(leaveResult1.isCompletedExceptionally());
        assertTrue(leaveResult2.isDone());
        assertFalse(leaveResult2.isCompletedExceptionally());

        // Subscription should have been updated only once with the first leave group.
        verify(subscriptionState, never()).assignFromSubscribed(Collections.emptySet());
    }
Members keep sending heartbeats while preparing to leave, so they could get fenced. This
tests ensures that if a member gets fenced while preparing to leave, it will directly
transition to UNSUBSCRIBE (rejoin as in regular fencing scenarios), and when the
LEAVING completes it remains UNSUBSCRIBED (no last heartbeat sent).
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareMembershipManagerTest.java