private ConsumerRebalanceListenerCallbackCompletedEvent mockFencedMemberStuckOnUserCallback(
        MembershipManagerImpl membershipManager,
        ConsumerRebalanceListenerInvoker invoker) {
        String topicName = "topic1";
        TopicPartition ownedPartition = new TopicPartition(topicName, 0);

        // Fence member and block waiting for onPartitionsLost callback to complete
        CounterConsumerRebalanceListener listener = new CounterConsumerRebalanceListener();
        when(subscriptionState.assignedPartitions()).thenReturn(Collections.singleton(ownedPartition));
        when(subscriptionState.hasAutoAssignedPartitions()).thenReturn(true);
        when(subscriptionState.rebalanceListener()).thenReturn(Optional.of(listener));
        when(commitRequestManager.autoCommitEnabled()).thenReturn(false);
        membershipManager.transitionToFenced();
        return performCallback(
            membershipManager,
            invoker,
            ConsumerRebalanceListenerMethodName.ON_PARTITIONS_LOST,
            topicPartitions(ownedPartition.topic(), ownedPartition.partition()),
            false
        );
    }
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/MembershipManagerImplTest.java