def send(self, node_id, request, wakeup=True, request_timeout_ms=None):
        """Send a request to a specific node. Bytes are placed on an
        internal per-connection send-queue. Actual network I/O will be
        triggered in a subsequent call to .poll()

        Arguments:
            node_id (int): destination node
            request (Struct): request object (not-encoded)

        Keyword Arguments:
            wakeup (bool, optional): optional flag to disable thread-wakeup.
            request_timeout_ms (int, optional): Provide custom timeout in milliseconds.
                If response is not processed before timeout, client will fail the
                request and close the connection.
                Default: None (uses value from client configuration)

        Raises:
            AssertionError: if node_id is not in current cluster metadata

        Returns:
            Future: resolves to Response struct or Error
        """
        conn = self._conns.get(node_id)
        if not conn or not self._can_send_request(node_id):
            self.maybe_connect(node_id, wakeup=wakeup)
            return Future().failure(Errors.NodeNotReadyError(node_id))

        # conn.send will queue the request internally
        # we will need to call send_pending_requests()
        # to trigger network I/O
        future = conn.send(request, blocking=False, request_timeout_ms=request_timeout_ms)
        if not future.is_done:
            self._sending.add(conn)

        # Wakeup signal is useful in case another thread is
        # blocked waiting for incoming network traffic while holding
        # the client lock in poll().
        if wakeup:
            self.wakeup()

        return future
Send a request to a specific node. Bytes are placed on an
        internal per-connection send-queue. Actual network I/O will be
        triggered in a subsequent call to .poll()

        Arguments:
            node_id (int): destination node
            request (Struct): request object (not-encoded)

        Keyword Arguments:
            wakeup (bool, optional): optional flag to disable thread-wakeup.
            request_timeout_ms (int, optional): Provide custom timeout in milliseconds.
                If response is not processed before timeout, client will fail the
                request and close the connection.
                Default: None (uses value from client configuration)

        Raises:
            AssertionError: if node_id is not in current cluster metadata

        Returns:
            Future: resolves to Response struct or Error
        
kafka/client_async.py