Kombu: SQS: JSONDecodeError when reading message from Amazon SQS predefined_queues

[januszm]

When reading messages from SQS using predefined_queues (messages come from SNS) I see the following JSON decode error in:

# kombu/transport/SQS.py", line 290
def _message_to_python(self, message, queue_name, queue):
    try:
        body = base64.b64decode(message['Body'].encode())
    except TypeError:
        body = message['Body'].encode()
    payload = loads(bytes_to_str(body)) # <<< HERE line 290

Afet dropping rdb.set_trace() around those lines I can see that the message looks of course just like regular AWS SNS/SQS message:

# simplified version for clarity
{
  'MessageId': 'abcd-efgh',
  'ReceiptHandle': 'abcdefgh==',
  'MD5OfBody': 'abc123',
  'Body': '''{\n  
    "Type" : "Notification",\n  
    "MessageId" : "abcd-efgh",\n  
    "TopicArn" : "arn:aws:sns:xxx",\n  
    "Subject" : "Amazon S3 Notification",\n  
    "Message" : "{\\"Records\\":[{\\"eventVersion\\":\\"2.1\\",\\"eventSource\\":\\"aws:s3\\",...}"
  }'''
}

The try: body = block works fine, it'll produce bytestring:

(Pdb) base64.b64decode(message['Body'].encode())
b'O*^6\x8ab~\'\x1c\xd6*\'1

but then it errors in line payload = loads(bytes_...

(Pdb) payload = loads(bytes_to_str(body))
*** json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)
(Pdb) bytes_to_str(body)
'O*^6�c~\'\x1b�*\

It looks like for messages coming from 'predefined_queues` (from other AWS publishers) we should simply:

loads(bytes_to_str(message['Body'].encode()))
# or even better
loads(message['Body'])

[januszm]

UPDATE: I found this answer at StackOverflow: https://stackoverflow.com/a/57389394/470512

The SQS support in Celery is only intended as a transport mechanism for Celery-specific messages. You cannot use Celery to consume arbitrary SQS messages.

Instead, I'd suggest writing a custom Django management command in which you poll your SQS queue using the boto3 library.

Yes, this can be a workaround, but in a sense it defeats the purpose of connecting to predefined queues. If you do not plan to handle arbitrary SQS messages in Celery, then it is worth mentioning this in the documentation or guides.

[januszm]

Continuing the previous message: maybe it might work and we could potentially use Celery/Kombu with any SQS messages if we could use custom message parsers or decoders such as:

def _message_to_python(self, message, queue_name, queue):
    if configured_message_parser(queue_name):
        payload = use custom_parser_here_(message)
    else:
        payload = default_messag_parser(message)
    # ...

This could be configured globally, for the whole app, or "per queue"

[qasimzee]

I would love this. This can literally open up celery to accept any kind of messages

[qasimzee]

@januszm what was your solution to this?

[thedrow]

We don't support arbitrary messages. We may consider it in a future version but right now it is not possible.

You should separate the queues you use for Celery and for other systems.

[mdettelson]

Hi there!

I'd like to reopen this discussion since I think it pertains to Kombu as a messaging library and not just how it integrates with Celery.

The way Kombu is described in the README is that it is "an idiomatic high-level interface for the AMQ protocol" and that it "Allows application authors to support several message server solutions by using pluggable transports." This seems like false advertising if there's the caveat of having to follow the Celery message format, which would be a pretty major caveat. My team was excited at the prospect of easily swapping some RMQ workers over to SQS without having to change the worker code drastically, but it seems like that won't be possible with Kombu.

@thedrow am I understanding this correctly? Apologies if I'm missing something here, I'm relatively new to Kombu :)

[mdettelson]

@auvipy tagging you since I think this is more of a philosophical question than an implementation one. Is Kombu meant to be used in isolation as a platform-agnostic generic messaging library?

[thedrow]

Yes. You can use Kombu in isolation and that's a very valid usecase.

[januszm]

@januszm what was your solution to this?

Unfortunately there was none, I stopped using Kombu in that project.