'Attempt to guess the version of a Kafka broker.
Note: It is possible that this method blocks longer than the
specified timeout. This can happen if the entire cluster
is down and the client enters a bootstrap backoff sleep.
This is only possible if node_id is None.
Returns: version tuple, i.e. (0, 10), (0, 9), (0, 8, 2), ...
Raises:
NodeNotReadyError (if node_id is provided)
NoBrokersAvailable (if node_id is None)
UnrecognizedBrokerVersion: please file bug if seen!
AssertionError (if strict=True): please file bug if seen!'

def check_version(self, node_id=None, timeout=2, strict=False):

    end = (time.time() + timeout)
    while (time.time() < end):
        try_node = (node_id or self.least_loaded_node())
        if (try_node is None):
            raise Errors.NoBrokersAvailable()
        self._maybe_connect(try_node)
        conn = self._conns[try_node]
        self._refresh_on_disconnects = False
        try:
            remaining = (end - time.time())
            version = conn.check_version(timeout=remaining, strict=strict)
            return version
        except Errors.NodeNotReadyError:
            if (node_id is not None):
                raise
        finally:
            self._refresh_on_disconnects = True
    else:
        raise Errors.NoBrokersAvailable()
