def send(self, request, blocking=True, request_timeout_ms=None):
        """Queue request for async network send, return Future()

        Arguments:
            request (Request): kafka protocol request object to send.

        Keyword Arguments:
            blocking (bool, optional): Whether to immediately send via
                blocking socket I/O. Default: True.
            request_timeout_ms: Custom timeout in milliseconds for request.
                Default: None (uses value from connection configuration)

        Returns: future
        """
        future = Future()
        if self.connecting():
            return future.failure(Errors.NodeNotReadyError(str(self)))
        elif not self.connected():
            return future.failure(Errors.KafkaConnectionError(str(self)))
        elif not self.can_send_more():
            # very small race here, but prefer it over breaking abstraction to check self._throttle_time
            if self.throttled():
                return future.failure(Errors.ThrottlingQuotaExceededError(str(self)))
            return future.failure(Errors.TooManyInFlightRequests(str(self)))
        return self._send(request, blocking=blocking, request_timeout_ms=request_timeout_ms)
Queue request for async network send, return Future()

        Arguments:
            request (Request): kafka protocol request object to send.

        Keyword Arguments:
            blocking (bool, optional): Whether to immediately send via
                blocking socket I/O. Default: True.
            request_timeout_ms: Custom timeout in milliseconds for request.
                Default: None (uses value from connection configuration)

        Returns: future
        
kafka/conn.py