Attempt to connect and return ConnectionState

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