docs: def connect(self):
          """Attempt to connect and return ConnectionState"""
          if self.state is ConnectionStates.DISCONNECTED and not self.blacked_out():
              self.last_attempt = time.time()
              next_lookup = self._next_afi_sockaddr()
              if not next_lookup:
                  self.close(Errors.KafkaConnectionError('DNS failure'))
                  return self.state
              else:
                  log.debug('%s: creating new socket', self)
                  assert self._sock is None
                  self._sock_afi, self._sock_addr = next_lookup
                  self._sock = socket.socket(self._sock_afi, socket.SOCK_STREAM)
  
              for option in self.config['socket_options']:
                  log.debug('%s: setting socket option %s', self, option)
                  self._sock.setsockopt(*option)
  
              self._sock.setblocking(False)
              self.state = ConnectionStates.CONNECTING
              self.config['state_change_callback'](self.node_id, self._sock, self)
              log.info('%s: connecting to %s:%d [%s %s]', self, self.host,
                       self.port, self._sock_addr, AFI_NAMES[self._sock_afi])
  
          if self.state is ConnectionStates.CONNECTING:
              # in non-blocking mode, use repeated calls to socket.connect_ex
              # to check connection status
              ret = None
              try:
                  ret = self._sock.connect_ex(self._sock_addr)
              except socket.error as err:
                  ret = err.errno
  
              # Connection succeeded
              if not ret or ret == errno.EISCONN:
                  log.debug('%s: established TCP connection', self)
  
                  if self.config['security_protocol'] in ('SSL', 'SASL_SSL'):
                      log.debug('%s: initiating SSL handshake', self)
                      self.state = ConnectionStates.HANDSHAKE
                      self.config['state_change_callback'](self.node_id, self._sock, self)
                      # _wrap_ssl can alter the connection state -- disconnects on failure
                      self._wrap_ssl()
  
                  elif self.config['security_protocol'] == 'SASL_PLAINTEXT':
                      log.debug('%s: initiating SASL authentication', self)
                      self.state = ConnectionStates.AUTHENTICATING
                      self.config['state_change_callback'](self.node_id, self._sock, self)
  
                  else:
                      # security_protocol PLAINTEXT
                      log.info('%s: Connection complete.', self)
                      self.state = ConnectionStates.CONNECTED
                      self._reset_reconnect_backoff()
                      self.config['state_change_callback'](self.node_id, self._sock, self)
  
              # Connection failed
              # WSAEINVAL == 10022, but errno.WSAEINVAL is not available on non-win systems
              elif ret not in (errno.EINPROGRESS, errno.EALREADY, errno.EWOULDBLOCK, 10022):
                  log.error('Connect attempt to %s returned error %s.'
                            ' Disconnecting.', self, ret)
                  errstr = errno.errorcode.get(ret, 'UNKNOWN')
                  self.close(Errors.KafkaConnectionError('{} {}'.format(ret, errstr)))
                  return self.state
  
              # Needs retry
              else:
                  pass
  
          if self.state is ConnectionStates.HANDSHAKE:
              if self._try_handshake():
                  log.debug('%s: completed SSL handshake.', self)
                  if self.config['security_protocol'] == 'SASL_SSL':
                      log.debug('%s: initiating SASL authentication', self)
                      self.state = ConnectionStates.AUTHENTICATING
                  else:
                      log.info('%s: Connection complete.', self)
                      self.state = ConnectionStates.CONNECTED
                      self._reset_reconnect_backoff()
                  self.config['state_change_callback'](self.node_id, self._sock, self)
  
          if self.state is ConnectionStates.AUTHENTICATING:
              assert self.config['security_protocol'] in ('SASL_PLAINTEXT', 'SASL_SSL')
              if self._try_authenticate():
                  # _try_authenticate has side-effects: possibly disconnected on socket errors
                  if self.state is ConnectionStates.AUTHENTICATING:
                      log.info('%s: Connection complete.', self)
                      self.state = ConnectionStates.CONNECTED
                      self._reset_reconnect_backoff()
                      self.config['state_change_callback'](self.node_id, self._sock, self)
  
          if self.state not in (ConnectionStates.CONNECTED,
                                ConnectionStates.DISCONNECTED):
              # Connection timed out
              request_timeout = self.config['request_timeout_ms'] / 1000.0
              if time.time() > request_timeout + self.last_attempt:
                  log.error('Connection attempt to %s timed out', self)
                  self.close(Errors.KafkaConnectionError('timeout'))
                  return self.state
  
          return self.state