private void testInternalTopicInclusion(boolean includeInternalTopics) {
        metadata = new ConsumerMetadata(0, 0, Long.MAX_VALUE, includeInternalTopics,
                false, subscriptions, new LogContext(), new ClusterResourceListeners());
        client = new MockClient(time, metadata);
        try (ConsumerCoordinator coordinator = buildCoordinator(rebalanceConfig, new Metrics(), assignors, false, subscriptions)) {
            subscriptions.subscribe(Pattern.compile(".*"), Optional.of(rebalanceListener));
            Node node = new Node(0, "localhost", 9999);
            MetadataResponse.PartitionMetadata partitionMetadata =
                new MetadataResponse.PartitionMetadata(Errors.NONE, new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 0),
                        Optional.of(node.id()), Optional.empty(), singletonList(node.id()), singletonList(node.id()),
                        singletonList(node.id()));
            MetadataResponse.TopicMetadata topicMetadata = new MetadataResponse.TopicMetadata(Errors.NONE,
                Topic.GROUP_METADATA_TOPIC_NAME, true, singletonList(partitionMetadata));

            client.updateMetadata(RequestTestUtils.metadataResponse(singletonList(node), "clusterId", node.id(),
                singletonList(topicMetadata)));
            coordinator.maybeUpdateSubscriptionMetadata();

            assertEquals(includeInternalTopics, subscriptions.subscription().contains(Topic.GROUP_METADATA_TOPIC_NAME));
        }
    }
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