[CRITICAL] Consumer encountered an unexpected error [rabbitmq]

[anapaulagomes]

This issue is similar to #266 but it comes from a different exception and a different broker (Rabbitmq).

What OS are you using?

Windows 10 + Ubuntu (WSL 2). Running in prod on Ubuntu 18.04.

What version of Dramatiq are you using?

dramatiq[rabbitmq,watch]==1.10.0

What did you do?

After getting some tasks in the queue, suddenly the exceptions are thrown.

What did you expect would happen?

No exceptions? 😅

What happened?

dramatiq_1  | [2021-03-28 14:23:55,963] [PID 12] [Thread-2] [dramatiq.worker.ConsumerThread(default)] [CRITICAL] Consumer encountered an unexpected error.
dramatiq_1  | Traceback (most recent call last):
dramatiq_1  |   File "/usr/local/lib/python3.8/site-packages/dramatiq/worker.py", line 259, in run
dramatiq_1  |     for message in self.consumer:
dramatiq_1  |   File "/usr/local/lib/python3.8/site-packages/dramatiq/brokers/rabbitmq.py", line 492, in __next__
dramatiq_1  |     message = Message.decode(body)
dramatiq_1  |   File "/usr/local/lib/python3.8/site-packages/dramatiq/message.py", line 96, in decode
dramatiq_1  |     return cls(**global_encoder.decode(data))
dramatiq_1  |   File "/usr/local/lib/python3.8/site-packages/dramatiq/encoder.py", line 52, in decode
dramatiq_1  |     return json.loads(data.decode("utf-8"))
dramatiq_1  |   File "/usr/local/lib/python3.8/json/__init__.py", line 357, in loads
dramatiq_1  |     return _default_decoder.decode(s)
dramatiq_1  |   File "/usr/local/lib/python3.8/json/decoder.py", line 337, in decode
dramatiq_1  |     obj, end = self.raw_decode(s, idx=_w(s, 0).end())
dramatiq_1  |   File "/usr/local/lib/python3.8/json/decoder.py", line 355, in raw_decode
dramatiq_1  |     raise JSONDecodeError("Expecting value", s, err.value) from None
dramatiq_1  | json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)
dramatiq_1  | [2021-03-28 14:23:55,974] [PID 11] [Thread-2] [dramatiq.worker.ConsumerThread(default)] [CRITICAL] Consumer encountered an unexpected error.
dramatiq_1  | Traceback (most recent call last):
dramatiq_1  |   File "/usr/local/lib/python3.8/site-packages/dramatiq/worker.py", line 259, in run
dramatiq_1  |     for message in self.consumer:
dramatiq_1  |   File "/usr/local/lib/python3.8/site-packages/dramatiq/brokers/rabbitmq.py", line 492, in __next__
dramatiq_1  |     message = Message.decode(body)
dramatiq_1  |   File "/usr/local/lib/python3.8/site-packages/dramatiq/message.py", line 96, in decode
dramatiq_1  |     return cls(**global_encoder.decode(data))
dramatiq_1  |   File "/usr/local/lib/python3.8/site-packages/dramatiq/encoder.py", line 52, in decode
dramatiq_1  |     return json.loads(data.decode("utf-8"))
dramatiq_1  |   File "/usr/local/lib/python3.8/json/__init__.py", line 357, in loads
dramatiq_1  |     return _default_decoder.decode(s)
dramatiq_1  |   File "/usr/local/lib/python3.8/json/decoder.py", line 337, in decode
dramatiq_1  |     obj, end = self.raw_decode(s, idx=_w(s, 0).end())
dramatiq_1  |   File "/usr/local/lib/python3.8/json/decoder.py", line 355, in raw_decode
dramatiq_1  |     raise JSONDecodeError("Expecting value", s, err.value) from None
dramatiq_1  | json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)
dramatiq_1  | [2021-03-28 14:23:55,977] [PID 12] [Thread-2] [dramatiq.worker.ConsumerThread(default)] [INFO] Restarting consumer in 3.00 seconds.
dramatiq_1  | [2021-03-28 14:23:55,978] [PID 12] [Thread-2] [dramatiq.brokers.rabbitmq._RabbitmqConsumer] [ERROR] Failed to wait for all callbacks to complete.  This can happen when the RabbitMQ server is suddenly restarted.

[Bogdanp]

Thanks for the report! The error makes it sound like either there was partial data in the message or the message was not JSON for whatever reason. I don't have any good ideas about why that might've happened (unless you're using a custom encoder somewhere and wires got crossed). #396 should at least help move past these in the future and it should help us figure out what the data looks like should the issue happen again.

[anapaulagomes]

Many thanks, @Bogdanp! 🥇