@Test
    public void testDelayedMetadataUsedToCompleteAssignment() {

        Uuid topicId1 = Uuid.randomUuid();
        String topic1 = "topic1";
        final TopicIdPartition topicId1Partition0 = new TopicIdPartition(topicId1, new TopicPartition(topic1, 0));

        Uuid topicId2 = Uuid.randomUuid();
        String topic2 = "topic2";
        final TopicIdPartition topicId2Partition0 = new TopicIdPartition(topicId2, new TopicPartition(topic2, 0));

        // Receive assignment with only topic1-0, entering STABLE state.
        ShareMembershipManager membershipManager =
                mockMemberSuccessfullyReceivesAndAcknowledgesAssignment(topicId1, topic1, Collections.singletonList(0));

        membershipManager.onHeartbeatRequestSent();

        assertEquals(MemberState.STABLE, membershipManager.state());
        when(subscriptionState.assignedPartitions()).thenReturn(getTopicPartitions(Collections.singleton(topicId1Partition0)));
        clearInvocations(membershipManager, subscriptionState);

        // New assignment adding a new topic2-0 (not in metadata).
        // No reconciliation triggered, because new topic in assignment is waiting for metadata.

        Map<Uuid, SortedSet<Integer>> newAssignment =
                mkMap(
                        mkEntry(topicId1, mkSortedSet(0)),
                        mkEntry(topicId2, mkSortedSet(0))
                );

        receiveAssignment(newAssignment, membershipManager);
        membershipManager.poll(time.milliseconds());

        verifyReconciliationNotTriggered(membershipManager);
        assertEquals(MemberState.RECONCILING, membershipManager.state());
        assertEquals(Collections.singleton(topicId2), membershipManager.topicsAwaitingReconciliation());
        verify(metadata).requestUpdate(anyBoolean());
        clearInvocations(membershipManager);

        // Metadata discovered for topic2. Should trigger reconciliation to complete the assignment,
        // with membership manager entering ACKNOWLEDGING state.

        Map<Uuid, String> fullTopicMetadata = mkMap(
                mkEntry(topicId1, topic1),
                mkEntry(topicId2, topic2)
        );
        when(metadata.topicNames()).thenReturn(fullTopicMetadata);

        membershipManager.poll(time.milliseconds());

        verifyReconciliationTriggeredAndCompleted(membershipManager, Arrays.asList(topicId1Partition0, topicId2Partition0));
    }
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