'Subscribe to a list of topics, or a topic regex pattern.
Partitions will be dynamically assigned via a group coordinator.
Topic subscriptions are not incremental: this list will replace the
current assignment (if there is one).
This method is incompatible with :meth:`~kafka.KafkaConsumer.assign`.
Arguments:
topics (list): List of topics for subscription.
pattern (str): Pattern to match available topics. You must provide
either topics or pattern, but not both.
listener (ConsumerRebalanceListener): Optionally include listener
callback, which will be called before and after each rebalance
operation.
As part of group management, the consumer will keep track of the
list of consumers that belong to a particular group and will
trigger a rebalance operation if one of the following events
trigger:
* Number of partitions change for any of the subscribed topics
* Topic is created or deleted
* An existing member of the consumer group dies
* A new member is added to the consumer group
When any of these events are triggered, the provided listener
will be invoked first to indicate that the consumer\'s assignment
has been revoked, and then again when the new assignment has
been received. Note that this listener will immediately override
any listener set in a previous call to subscribe. It is
guaranteed, however, that the partitions revoked/assigned
through this interface are from topics subscribed in this call.
Raises:
IllegalStateError: If called after previously calling
:meth:`~kafka.KafkaConsumer.assign`.
AssertionError: If neither topics or pattern is provided.
TypeError: If listener is not a ConsumerRebalanceListener.'

def subscribe(self, topics=(), pattern=None, listener=None):

    self._subscription.subscribe(topics=topics, pattern=pattern, listener=listener)
    if (pattern is not None):
        self._client.cluster.need_all_topic_metadata = True
        self._client.set_topics([])
        self._client.cluster.request_update()
        log.debug('Subscribed    to    topic    pattern:    %s', pattern)
    else:
        self._client.cluster.need_all_topic_metadata = False
        self._client.set_topics(self._subscription.group_subscription())
        log.debug('Subscribed    to    topic(s):    %s', topics)
