'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):

    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)
