@Test
    public void testFencingWhenStateIsPrepareLeaving() {
        MembershipManagerImpl membershipManager = createMemberInStableState();
        ConsumerRebalanceListenerInvoker invoker = consumerRebalanceListenerInvoker();
        ConsumerRebalanceListenerCallbackCompletedEvent callbackEvent =
            mockPrepareLeavingStuckOnUserCallback(membershipManager, invoker);
        assertEquals(MemberState.PREPARE_LEAVING, membershipManager.state());

        // Get fenced while preparing to leave the group. Member should ignore the fence
        // (no callbacks or rejoin) and should transition to UNSUBSCRIBED to
        // effectively stop sending heartbeats.
        clearInvocations(subscriptionState);
        membershipManager.transitionToFenced();
        testFenceIsNoOp(membershipManager);
        assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());

        // When PREPARE_LEAVING completes, the member should remain UNSUBSCRIBED (no last HB sent
        // because member is already out of the group in the broker).
        completeCallback(callbackEvent, membershipManager);
        assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());
        assertEquals(ConsumerGroupHeartbeatRequest.LEAVE_GROUP_MEMBER_EPOCH, membershipManager.memberEpoch());
        verify(membershipManager).notifyEpochChange(Optional.empty(), Optional.empty());
        assertTrue(membershipManager.shouldSkipHeartbeat());
    }
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 (no callbacks triggered or rejoin as in regular fencing
scenarios), and when the PREPARE_LEAVING completes it remains UNSUBSCRIBED (no last
heartbeat sent).
clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java