def check_version(self, node_id=None, timeout=None, **kwargs):
        """Attempt to guess the version of a Kafka broker.

        Keyword Arguments:
            node_id (str, optional): Broker node id from cluster metadata. If None, attempts
                to connect to any available broker until version is identified.
                Default: None
            timeout (num, optional): Maximum time in seconds to try to check broker version.
                If unable to identify version before timeout, raise error (see below).
                Default: api_version_auto_timeout_ms / 1000

        Returns: version tuple, i.e. (3, 9), (2, 0), (0, 10, 2) etc

        Raises:
            NodeNotReadyError (if node_id is provided)
            NoBrokersAvailable (if node_id is None)
        """
        timeout = timeout or (self.config['api_version_auto_timeout_ms'] / 1000)
        with self._lock:
            end = time.time() + timeout
            while time.time() < end:
                time_remaining = max(end - time.time(), 0)
                if node_id is not None and self.connection_delay(node_id) > 0:
                    sleep_time = min(time_remaining, self.connection_delay(node_id) / 1000.0)
                    if sleep_time > 0:
                        time.sleep(sleep_time)
                    continue
                try_node = node_id or self.least_loaded_node()
                if try_node is None:
                    sleep_time = min(time_remaining,  self.least_loaded_node_refresh_ms() / 1000.0)
                    if sleep_time > 0:
                        log.warning('No node available during check_version; sleeping %.2f secs', sleep_time)
                        time.sleep(sleep_time)
                    continue
                log.debug('Attempting to check version with node %s', try_node)
                if not self._init_connect(try_node):
                    if try_node == node_id:
                        raise Errors.NodeNotReadyError("Connection failed to %s" % node_id)
                    else:
                        continue
                conn = self._conns[try_node]

                while conn.connecting() and time.time() < end:
                    timeout_ms = min((end - time.time()) * 1000, 200)
                    self.poll(timeout_ms=timeout_ms)

                if conn._api_version is not None:
                    return conn._api_version
                else:
                    log.debug('Failed to identify api_version after connection attempt to %s', conn)

            # Timeout
            else:
                if node_id is not None:
                    raise Errors.NodeNotReadyError(node_id)
                else:
                    raise Errors.NoBrokersAvailable()
Attempt to guess the version of a Kafka broker.

        Keyword Arguments:
            node_id (str, optional): Broker node id from cluster metadata. If None, attempts
                to connect to any available broker until version is identified.
                Default: None
            timeout (num, optional): Maximum time in seconds to try to check broker version.
                If unable to identify version before timeout, raise error (see below).
                Default: api_version_auto_timeout_ms / 1000

        Returns: version tuple, i.e. (3, 9), (2, 0), (0, 10, 2) etc

        Raises:
            NodeNotReadyError (if node_id is provided)
            NoBrokersAvailable (if node_id is None)
        
kafka/client_async.py