@Override
    public void subscribe(Pattern pattern, ConsumerRebalanceListener listener) {
        if (listener == null)
            throw new IllegalArgumentException("RebalanceListener cannot be null");

        subscribeInternal(pattern, Optional.of(listener));
    }
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