private void subscribeInternal(Collection<String> topics, Optional<ConsumerRebalanceListener> listener) {
        acquireAndEnsureOpen();
        try {
            maybeThrowInvalidGroupIdException();
            if (topics == null)
                throw new IllegalArgumentException("Topic collection to subscribe to cannot be null");
            if (topics.isEmpty()) {
                // treat subscribing to empty topic list as the same as unsubscribing
                this.unsubscribe();
            } else {
                for (String topic : topics) {
                    if (isBlank(topic))
                        throw new IllegalArgumentException("Topic collection to subscribe to cannot contain null or empty topic");
                }

                throwIfNoAssignorsConfigured();

                // Clear the buffered data which are not a part of newly assigned topics
                final Set<TopicPartition> currentTopicPartitions = new HashSet<>();

                for (TopicPartition tp : subscriptions.assignedPartitions()) {
                    if (topics.contains(tp.topic()))
                        currentTopicPartitions.add(tp);
                }

                fetcher.clearBufferedDataForUnassignedPartitions(currentTopicPartitions);

                log.info("Subscribed to topic(s): {}", String.join(", ", topics));
                if (this.subscriptions.subscribe(new HashSet<>(topics), listener))
                    metadata.requestUpdateForNewTopics();
            }
        } finally {
            release();
        }
    }
Internal helper method for {@link #subscribe(Collection)} and
{@link #subscribe(Collection, ConsumerRebalanceListener)}
<p>
Subscribe to the given list of topics to get dynamically assigned partitions.
<b>Topic subscriptions are not incremental. This list will replace the current
assignment (if there is one).</b> It is not possible to combine topic subscription with group management
with manual partition assignment through {@link #assign(Collection)}.

If the given list of topics is empty, it is treated the same as {@link #unsubscribe()}.

<p>
@param topics The list of topics to subscribe to
@param listener {@link Optional} listener instance to get notifications on partition assignment/revocation
                for the subscribed topics
@throws IllegalArgumentException If topics is null or contains null or empty elements
@throws IllegalStateException If {@code subscribe()} is called previously with pattern, or assign is called
                              previously (without a subsequent call to {@link #unsubscribe()}), or if not
                              configured at-least one partition assignment strategy
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ClassicKafkaConsumer.java