def _init_connect(self, node_id):
        """Idempotent non-blocking connection attempt to the given node id.

        Returns True if connection object exists and is connected / connecting
        """
        with self._lock:
            conn = self._conns.get(node_id)

            # Check if existing connection should be recreated because host/port changed
            if conn is not None and self._should_recycle_connection(conn):
                self._conns.pop(node_id).close()
                conn = None

            if conn is None:
                broker = self.cluster.broker_metadata(node_id)
                if broker is None:
                    log.debug('Broker id %s not in current metadata', node_id)
                    return False

                log.debug("Initiating connection to node %s at %s:%s",
                          node_id, broker.host, broker.port)
                host, port, afi = get_ip_port_afi(broker.host)
                cb = WeakMethod(self._conn_state_change)
                conn = BrokerConnection(host, broker.port, afi,
                                        state_change_callback=cb,
                                        node_id=node_id,
                                        **self.config)
                self._conns[node_id] = conn

            if conn.disconnected():
                conn.connect()
            return not conn.disconnected()
Idempotent non-blocking connection attempt to the given node id.

        Returns True if connection object exists and is connected / connecting
        
kafka/client_async.py