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

def _wait_on_metadata(self, topic, max_wait):
        """
        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
        """
        # 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)