@ParameterizedTest
    @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
    public void testChangingRegexSubscription(GroupProtocol groupProtocol) {
        String otherTopic = "other";
        TopicPartition otherTopicPartition = new TopicPartition(otherTopic, 0);

        ConsumerMetadata metadata = createMetadata(subscription);
        MockClient client = new MockClient(time, metadata);

        Map<String, Integer> partitionCounts = new HashMap<>();
        partitionCounts.put(topic, 1);
        partitionCounts.put(otherTopic, 1);
        topicIds.put(otherTopic, Uuid.randomUuid());
        initMetadata(client, partitionCounts);
        Node node = metadata.fetch().nodes().get(0);

        consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false, groupInstanceId);

        Node coordinator = prepareRebalance(client, node, singleton(topic), assignor, singletonList(tp0), null);
        consumer.subscribe(Pattern.compile(topic), getConsumerRebalanceListener(consumer));

        consumer.updateAssignmentMetadataIfNeeded(time.timer(Long.MAX_VALUE));
        consumer.poll(Duration.ZERO);

        assertEquals(singleton(topic), consumer.subscription());

        consumer.subscribe(Pattern.compile(otherTopic), getConsumerRebalanceListener(consumer));

        client.prepareMetadataUpdate(RequestTestUtils.metadataUpdateWithIds(1, partitionCounts, topicIds));
        prepareRebalance(client, node, singleton(otherTopic), assignor, singletonList(otherTopicPartition), coordinator);
        consumer.poll(Duration.ZERO);

        assertEquals(singleton(otherTopic), consumer.subscription());
    }
Note to future authors in this class. If you close the consumer, close with DURATION.ZERO to reduce the duration of
the test.
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java