docs: def check_version(self, timeout=2, strict=False, topics=[]):
          """Attempt to guess the broker version.
  
          Note: This is a blocking call.
  
          Returns: version tuple, i.e. (0, 10), (0, 9), (0, 8, 2), ...
          """
          timeout_at = time.time() + timeout
          log.info('Probing node %s broker version', self.node_id)
          # Monkeypatch some connection configurations to avoid timeouts
          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]
  
          # kafka kills the connection when it doesn't recognize an API request
          # so we can send a test request and then follow immediately with a
          # vanilla MetadataRequest. If the server did not recognize the first
          # request, both will be failed with a ConnectionError that wraps
          # socket.error (32, 54, or 104)
          from kafka.protocol.admin import ApiVersionRequest, ListGroupsRequest
          from kafka.protocol.commit import OffsetFetchRequest, GroupCoordinatorRequest
  
          test_cases = [
              # All cases starting from 0.10 will be based on ApiVersionResponse
              ((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](topics)),
          ]
  
          for version, request in test_cases:
              if not self.connect_blocking(timeout_at - time.time()):
                  raise Errors.NodeNotReadyError()
              f = self.send(request)
              # HACK: sleeping to wait for socket to send bytes
              time.sleep(0.1)
              # when broker receives an unrecognized request API
              # it abruptly closes our socket.
              # so we attempt to send a second request immediately
              # that we believe it will definitely recognize (metadata)
              # the attempt to write to a disconnected socket should
              # immediately fail and allow us to infer that the prior
              # request was unrecognized
              mr = self.send(MetadataRequest[0](topics))
  
              selector = self.config['selector']()
              selector.register(self._sock, selectors.EVENT_READ)
              while not (f.is_done and mr.is_done):
                  selector.select(1)
                  for response, future in self.recv():
                      future.success(response)
              selector.close()
  
              if f.succeeded():
                  if isinstance(request, ApiVersionRequest[0]):
                      # Starting from 0.10 kafka broker we determine version
                      # by looking at ApiVersionResponse
                      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
  
              # Only enable strict checking to verify that we understand failure
              # modes. For most users, the fact that the request failed should be
              # enough to rule out a particular broker version.
              if strict:
                  # If the socket flush hack did not work (which should force the
                  # connection to close and fail all pending requests), then we
                  # get a basic Request Timeout. This is not ideal, but we'll deal
                  if isinstance(f.exception, Errors.RequestTimedOutError):
                      pass
  
                  # 0.9 brokers do not close the socket on unrecognized api
                  # requests (bug...). In this case we expect to see a correlation
                  # id mismatch
                  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()
  
          for key in stashed:
              self.config[key] = stashed[key]
          return version