docs: def load_metadata_for_topics(self, *topics, **kwargs):
          """Fetch broker and topic-partition metadata from the server.
  
          Updates internal data: broker list, topic/partition list, and
          topic/partition -> 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).
          """
          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:
              # Errors expected for new 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
  
                  # Populate topics_to_brokers dict
                  topic_part = TopicPartition(topic, partition)
  
                  # Check for partition errors
                  if error:
                      error_type = kafka.errors.kafka_errors.get(error, UnknownError)
  
                      # If No Leader, topics_to_brokers topic_partition -> None
                      if error_type is 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
                      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 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 don't 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, None
                      )