def _maybe_refresh_metadata(self, wakeup=False):
        """Send a metadata request if needed.

        Returns:
            float: milliseconds until next refresh
        """
        ttl = self.cluster.ttl()
        wait_for_in_progress_ms = self.config['request_timeout_ms'] if self._metadata_refresh_in_progress else 0
        metadata_timeout = max(ttl, wait_for_in_progress_ms)

        if metadata_timeout > 0:
            return metadata_timeout

        # Beware that the behavior of this method and the computation of
        # timeouts for poll() are highly dependent on the behavior of
        # least_loaded_node()
        node_id = self.least_loaded_node()
        if node_id is None:
            next_connect_ms = self.least_loaded_node_refresh_ms()
            log.debug("Give up sending metadata request since no node is available. (reconnect delay %d ms)", next_connect_ms)
            return next_connect_ms

        if not self._can_send_request(node_id):
            # If there's any connection establishment underway, wait until it completes. This prevents
            # the client from unnecessarily connecting to additional nodes while a previous connection
            # attempt has not been completed.
            if self._connecting:
                return float('inf')

            elif self._can_connect(node_id):
                log.debug("Initializing connection to node %s for metadata request", node_id)
                self._connecting.add(node_id)
                if not self._init_connect(node_id):
                    if node_id in self._connecting:
                        self._connecting.remove(node_id)
                    # Connection attempt failed immediately, need to retry with a different node
                    return self.config['reconnect_backoff_ms']
            else:
                # Existing connection throttled or max in flight requests.
                return self.throttle_delay(node_id) or self.config['request_timeout_ms']

        # Recheck node_id in case we were able to connect immediately above
        if self._can_send_request(node_id):
            topics = list(self._topics)
            if not topics and self.cluster.is_bootstrap(node_id):
                topics = list(self.config['bootstrap_topics_filter'])

            api_version = self.api_version(MetadataRequest, max_version=7)
            if self.cluster.need_all_topic_metadata:
                topics = MetadataRequest[api_version].ALL_TOPICS
            elif not topics:
                topics = MetadataRequest[api_version].NO_TOPICS
            if api_version >= 4:
                request = MetadataRequest[api_version](topics, self.config['allow_auto_create_topics'])
            else:
                request = MetadataRequest[api_version](topics)
            log.debug("Sending metadata request %s to node %s", request, node_id)
            future = self.send(node_id, request, wakeup=wakeup)
            future.add_callback(self.cluster.update_metadata)
            future.add_errback(self.cluster.failed_update)

            self._metadata_refresh_in_progress = True
            def refresh_done(val_or_error):
                self._metadata_refresh_in_progress = False
            future.add_callback(refresh_done)
            future.add_errback(refresh_done)
            return self.config['request_timeout_ms']

        # Should only get here if still connecting
        if self._connecting:
            return float('inf')
        else:
            return self.config['reconnect_backoff_ms']
Send a metadata request if needed.

        Returns:
            float: milliseconds until next refresh
        
kafka/client_async.py