private void unavailableTopicTest(boolean patternSubscribe, Set<String> unavailableTopicsInLastMetadata) {
        if (patternSubscribe)
            subscriptions.subscribe(Pattern.compile("test.*"), Optional.of(rebalanceListener));
        else
            subscriptions.subscribe(singleton(topic1), Optional.of(rebalanceListener));

        client.prepareMetadataUpdate(RequestTestUtils.metadataUpdateWith("kafka-cluster", 1,
                Collections.singletonMap(topic1, Errors.UNKNOWN_TOPIC_OR_PARTITION), Collections.emptyMap()));

        client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
        coordinator.ensureCoordinatorReady(time.timer(Long.MAX_VALUE));

        Map<String, List<String>> memberSubscriptions = singletonMap(consumerId, singletonList(topic1));
        partitionAssignor.prepare(Collections.emptyMap());

        client.prepareResponse(joinGroupLeaderResponse(1, consumerId, memberSubscriptions, Errors.NONE));
        client.prepareResponse(syncGroupResponse(Collections.emptyList(), Errors.NONE));
        coordinator.poll(time.timer(Long.MAX_VALUE));
        assertFalse(coordinator.rejoinNeededOrPending());
        // callback not triggered since there's nothing to be assigned
        assertEquals(Collections.emptySet(), rebalanceListener.assigned);
        assertTrue(metadata.updateRequested(), "Metadata refresh not requested for unavailable partitions");

        Map<String, Errors> topicErrors = new HashMap<>();
        for (String topic : unavailableTopicsInLastMetadata)
            topicErrors.put(topic, Errors.UNKNOWN_TOPIC_OR_PARTITION);

        client.prepareMetadataUpdate(RequestTestUtils.metadataUpdateWith("kafka-cluster", 1,
                topicErrors, singletonMap(topic1, 1)));

        consumerClient.poll(time.timer(0));
        client.prepareResponse(joinGroupLeaderResponse(2, consumerId, memberSubscriptions, Errors.NONE));
        client.prepareResponse(syncGroupResponse(singletonList(t1p), Errors.NONE));
        coordinator.poll(time.timer(Long.MAX_VALUE));

        assertFalse(metadata.updateRequested(), "Metadata refresh requested unnecessarily");
        assertFalse(coordinator.rejoinNeededOrPending());
        assertEquals(singleton(t1p), rebalanceListener.assigned);
    }
Verifies that subscription change updates SubscriptionState correctly even after JoinGroup failures
that don't re-invoke onJoinPrepare.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java