'Attempt to guess the broker version.
Note: This is a blocking call.
Returns: version tuple, i.e. (0, 10), (0, 9), (0, 8, 2), ...'

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

    override_config = {'request_timeout_ms': (timeout * 1000), 'max_in_flight_requests_per_connection': 5}
    stashed = {}
    for key in override_config:
        stashed[key] = self.config[key]
        self.config[key] = override_config[key]
    from .protocol.admin import ApiVersionRequest, ListGroupsRequest
    from .protocol.commit import OffsetFetchRequest, GroupCoordinatorRequest
    from logging import Filter
    class ConnFilter(Filter, ):
        def filter(self, record):
            if (record.funcName == 'check_version'):
                return True
            return False
    log_filter = ConnFilter()
    log.addFilter(log_filter)
    test_cases = [((0, 10), ApiVersionRequest[0]()), ((0, 9), ListGroupsRequest[0]()), ((0, 8, 2), GroupCoordinatorRequest[0]('kafka-python-default-group')), ((0, 8, 1), OffsetFetchRequest[0]('kafka-python-default-group', [])), ((0, 8, 0), MetadataRequest[0]([]))]
    def connect():
        self.connect()
        if self.connected():
            return
        timeout_at = (time.time() + timeout)
        while ((time.time() < timeout_at) and self.connecting()):
            if (self.connect() is ConnectionStates.CONNECTED):
                return
            time.sleep(0.05)
        raise Errors.NodeNotReadyError()
    for (version, request) in test_cases:
        connect()
        f = self.send(request)
        time.sleep(0.1)
        mr = self.send(MetadataRequest[0]([]))
        if self._sock:
            self._sock.setblocking(True)
        while (not (f.is_done and mr.is_done)):
            self.recv()
        if self._sock:
            self._sock.setblocking(False)
        if f.succeeded():
            if isinstance(request, ApiVersionRequest[0]):
                api_versions = self._handle_api_version_response(f.value)
                version = self._infer_broker_version_from_api_versions(api_versions)
            log.info('Broker    version    identifed    as    %s', '.'.join(map(str, version)))
            log.info('Set    configuration    api_version=%s    to    skip    auto    check_version    requests    on    startup', version)
            break
        if strict:
            if isinstance(f.exception, Errors.RequestTimedOutError):
                pass
            elif (isinstance(f.exception, Errors.CorrelationIdError) and (version == (0, 10))):
                pass
            elif six.PY2:
                assert isinstance(f.exception.args[0], socket.error)
                assert (f.exception.args[0].errno in (32, 54, 104))
            else:
                assert isinstance(f.exception.args[0], ConnectionError)
        log.info('Broker    is    not    v%s    --    it    did    not    recognize    %s', version, request.__class__.__name__)
    else:
        raise Errors.UnrecognizedBrokerVersion()
    log.removeFilter(log_filter)
    for key in stashed:
        self.config[key] = stashed[key]
    return version
