code: def _wait_on_metadata(self, topic, max_wait):
          
          # add topic to metadata topic list if it is not there already.
          self._sender.add_topic(topic)
          begin = time.time()
          elapsed = 0.0
          metadata_event = None
          while True:
              partitions = self._metadata.partitions_for_topic(topic)
              if partitions is not None:
                  return partitions
  
              if not metadata_event:
                  metadata_event = threading.Event()
  
              log.debug("Requesting metadata update for topic %s", topic)
  
              metadata_event.clear()
              future = self._metadata.request_update()
              future.add_both(lambda e, *args: e.set(), metadata_event)
              self._sender.wakeup()
              metadata_event.wait(max_wait - elapsed)
              elapsed = time.time() - begin
              if not metadata_event.is_set():
                  raise Errors.KafkaTimeoutError(
                      "Failed to update metadata after %.1f secs." % (max_wait,))
              elif topic in self._metadata.unauthorized_topics:
                  raise Errors.TopicAuthorizationFailedError(topic)
              else:
                  log.debug("_wait_on_metadata woke after %s secs.", elapsed)
docs: Wait for cluster metadata including partitions for the given topic to
          be available.
  
          Arguments:
              topic (str): topic we want metadata for
              max_wait (float): maximum time in secs for waiting on the metadata
  
          Returns:
              set: partition ids for the topic
  
          Raises:
              KafkaTimeoutError: if partitions for topic were not obtained before
                  specified max_wait timeout