@Test
    public void testLeaveGroupWhenMemberOwnsAssignment() {
        Uuid topicId = Uuid.randomUuid();
        String topicName = "topic1";
        ShareMembershipManager membershipManager = createMembershipManagerJoiningGroup();
        mockOwnedPartitionAndAssignmentReceived(membershipManager, topicId, topicName, Collections.emptyList());

        receiveAssignment(topicId, Arrays.asList(0, 1), membershipManager);

        verifyReconciliationNotTriggered(membershipManager);
        membershipManager.poll(time.milliseconds());

        List<TopicIdPartition> assignedPartitions = Arrays.asList(
                new TopicIdPartition(topicId, new TopicPartition(topicName, 0)),
                new TopicIdPartition(topicId, new TopicPartition(topicName, 1)));
        verifyReconciliationTriggeredAndCompleted(membershipManager, assignedPartitions);

        assertEquals(1, membershipManager.currentAssignment().size());

        testLeaveGroupReleasesAssignmentAndResetsEpochToSendLeaveGroup(membershipManager);
    }
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