docs: def load_metadata_for_topics(self, *topics):
          """
          Fetch broker and topic-partition metadata from the server,
          and update internal data:
          broker list, topic/partition list, and topic/parition -> broker map
  
          This method should be called after receiving any error
  
          Arguments:
              *topics (optional): If a list of topics is provided,
                  the metadata refresh will be limited to the specified topics only.
  
          Exceptions:
          ----------
          If the broker is configured to not auto-create topics,
          expect UnknownTopicOrPartitionError for topics that don't exist
  
          If the broker is configured to auto-create topics,
          expect LeaderNotAvailableError for new topics
          until partitions have been initialized.
  
          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)
          """
          topics = [kafka_bytestring(t) for t in topics]
  
          if topics:
              for topic in topics:
                  self.reset_topic_metadata(topic)
          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', resp.topics)
  
          self.brokers = dict([(broker.nodeId, broker)
                               for broker in resp.brokers])
  
          for topic_metadata in resp.topics:
              topic = topic_metadata.topic
              partitions = topic_metadata.partitions
  
              # Errors expected for new topics
              try:
                  kafka_common.check_error(topic_metadata)
              except (UnknownTopicOrPartitionError, LeaderNotAvailableError) as e:
  
                  # Raise if the topic was passed in explicitly
                  if topic in topics:
                      raise
  
                  # Otherwise, just log a warning
                  log.error('Error loading topic metadata for %s: %s', topic, type(e))
                  continue
  
              self.topic_partitions[topic] = {}
              for partition_metadata in partitions:
                  partition = partition_metadata.partition
                  leader = partition_metadata.leader
  
                  self.topic_partitions[topic][partition] = partition_metadata
  
                  # Populate topics_to_brokers dict
                  topic_part = TopicAndPartition(topic, partition)
  
                  # Check for partition errors
                  try:
                      kafka_common.check_error(partition_metadata)
  
                  # If No Leader, topics_to_brokers topic_partition -> None
                  except LeaderNotAvailableError:
                      log.error('No leader for topic %s partition %d', topic, partition)
                      self.topics_to_brokers[topic_part] = None
                      continue
                  # If one of the replicas is unavailable -- ignore
                  # this error code is provided for admin purposes only
                  # we never talk to replicas, only the leader
                  except ReplicaNotAvailableError:
                      log.debug('Some (non-leader) replicas not available for topic %s partition %d',
                                topic, partition)
  
                  # If Known Broker, topic_partition -> BrokerMetadata
                  if leader in self.brokers:
                      self.topics_to_brokers[topic_part] = self.brokers[leader]
  
                  # If Unknown Broker, fake BrokerMetadata so we dont lose the id
                  # (not sure how this could happen. server could be in bad state)
                  else:
                      self.topics_to_brokers[topic_part] = BrokerMetadata(
                          leader, None, None
                      )