'Fetch broker and topic-partition metadata from the server.
Updates internal data: broker list, topic/partition list, and
topic/parition -> broker map. This method should be called after
receiving any error.
Note: Exceptions *will not* be raised in a full refresh (i.e. no topic
list). In this case, error codes will be logged as errors.
Partition-level errors will also not be raised here (a single partition
w/o a leader, for example).
Arguments:
*topics (optional): If a list of topics is provided,
the metadata refresh will be limited to the specified topics
only.
ignore_leadernotavailable (bool): suppress LeaderNotAvailableError
so that metadata is loaded correctly during auto-create.
Default: False.
Raises:
UnknownTopicOrPartitionError: Raised for topics that do not exist,
unless the broker is configured to auto-create topics.
LeaderNotAvailableError: Raised for topics that do not exist yet,
when the broker is configured to auto-create topics. Retry
after a short backoff (topics/partitions are initializing).'

def load_metadata_for_topics(self, *topics, **kwargs):

    if ('ignore_leadernotavailable' in kwargs):
        ignore_leadernotavailable = kwargs['ignore_leadernotavailable']
    else:
        ignore_leadernotavailable = False
    if topics:
        self.reset_topic_metadata(*topics)
    else:
        self.reset_all_metadata()
    resp = self.send_metadata_request(topics)
    log.debug('Updating    broker    metadata:    %s', resp.brokers)
    log.debug('Updating    topic    metadata:    %s', [topic for (_, topic, _) in resp.topics])
    self.brokers = dict([(nodeId, BrokerMetadata(nodeId, host, port, None)) for (nodeId, host, port) in resp.brokers])
    for (error, topic, partitions) in resp.topics:
        if error:
            error_type = kafka.errors.kafka_errors.get(error, UnknownError)
            if (error_type in (UnknownTopicOrPartitionError, LeaderNotAvailableError)):
                log.error('Error    loading    topic    metadata    for    %s:    %s    (%s)', topic, error_type, error)
                if (topic not in topics):
                    continue
                elif ((error_type is LeaderNotAvailableError) and ignore_leadernotavailable):
                    continue
            raise error_type(topic)
        self.topic_partitions[topic] = {}
        for (error, partition, leader, _, _) in partitions:
            self.topic_partitions[topic][partition] = leader
            topic_part = TopicPartition(topic, partition)
            if error:
                error_type = kafka.errors.kafka_errors.get(error, UnknownError)
                if (error_type is LeaderNotAvailableError):
                    log.error('No    leader    for    topic    %s    partition    %d', topic, partition)
                    self.topics_to_brokers[topic_part] = None
                    continue
                elif (error_type is ReplicaNotAvailableError):
                    log.debug('Some    (non-leader)    replicas    not    available    for    topic    %s    partition    %d', topic, partition)
                else:
                    raise error_type(topic_part)
            if (leader in self.brokers):
                self.topics_to_brokers[topic_part] = self.brokers[leader]
            else:
                self.topics_to_brokers[topic_part] = BrokerMetadata(leader, None, None, None)
