Amazon MQ rejecting connections from aio_pika, has anyone succeeded to use TaskIQ on AWS?

[devraj]

Hi all,

This is a bit particular to AWS, we are trying to use TaskIQ with Amazon MQ which requires us to connect via this URL scheme (where we must use SSL to connect):

amqp+ssl://b-hostname-1.mq.ap-southeast-2.amazonaws.com:5671

which produces the following error when the broker starts up (in a FastAPI application):

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/local/lib/python3.11/asyncio/runners.py", line 190, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/asyncio/base_events.py", line 653, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/aio_pika/robust_connection.py", line 331, in connect_robust
    await connection.connect(timeout=timeout)
  File "/usr/local/lib/python3.11/site-packages/aio_pika/robust_connection.py", line 177, in connect
    await self.__fail_fast_future
  File "/usr/local/lib/python3.11/site-packages/aio_pika/robust_connection.py", line 132, in __connection_factory
    await Connection.connect(self, self.__connect_timeout)
  File "/usr/local/lib/python3.11/site-packages/aio_pika/connection.py", line 100, in connect
    self.transport = await UnderlayConnection.connect(
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/aio_pika/abc.py", line 664, in connect
    connection = await cls.make_connection(
                 ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/aio_pika/abc.py", line 652, in make_connection
    connection: aiormq.abc.AbstractConnection = await asyncio.wait_for(
                                                ^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/asyncio/tasks.py", line 442, in wait_for
    return await fut
           ^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/aiormq/connection.py", line 913, in connect
    await connection.connect(client_properties or {})
  File "/usr/local/lib/python3.11/site-packages/aiormq/base.py", line 164, in wrap
    return await self.create_task(func(self, *args, **kwargs))
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/aiormq/abc.py", line 44, in __inner
    return await self.task
           ^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/aiormq/connection.py", line 465, in connect
    _, _, frame = await frame_receiver.get_frame()
                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/aiormq/connection.py", line 192, in get_frame
    raise AMQPSyntaxError
pamqp.exceptions.AMQPSyntaxError
>>> 

trying to debug it, I found that this to do with aio_pika and hence got it down initialise a connection manually:

import asyncio
import aio_pika
c = asyncio.run(aio_pika.connect_robust('amqps://user:[email protected]:5671/'))

and I did get the same error.

Knowing that this is unrelated to taskiq, I was reaching out to see if there was anyone else using TaskIQ against AmazonMQ and if there were any insights you could share with me?

I did find this article outlining a Python example.

I also wondered if it's possible to use other protocols to connect to the queue like STOMP in a TaskIQ context?

Thank you for your time.

[s3rius]

Since I see syntax error for protocol, I guess it sends you encrypted data and you have to provide itwith certificate in order to decrypt data.

https://aio-pika.readthedocs.io/en/latest/apidoc.html#aio_pika.connect_robust

Here's a method reference, and, as you can see, tls is set to False by default. Since in the example you've sent they use SSLContext and certificates.

[s3rius]

About other protocols, actually we support different protocols, but only for specific queues. As I remember, some guy tried to implement his own lib for supporting amazonMQ or SQS, but he didn't published anything yet.

I can try add new brokers for SQS and AmazonMQ.

[devraj]

So turned out that AmazonMQ defaults to using ActiveMQ which exposes the endpoint as amqp+ssl:// which would be incompatible with amqps, they do do have a RabbitMQ offering which uses the amqp protocol, and changing to this makes everything work as expected.

Hopefully this is helpful for anyone out there using TaskIQ on AWS, that you should use an RabbitMQ instance not an ActiveMQ.