Celery SQS Consumer Crashes with JSONDecodeError on Invalid / Non-Celery Messages

[emrementese]

Problem Summary

There is a critical reliability issue in Celery/Kombu when using the SQS transport.

If any message reaches the SQS queue that is not a valid Celery-formatted JSON payload (for example: AWS Console test message, a misconfigured service, a typo, or any non-Celery producer), Kombu attempts to execute:

payload = loads(bytes_to_str(body))

This immediately triggers a JSONDecodeError, which propagates up the stack and ultimately causes the following:

Unrecoverable error: JSONDecodeError(...)
-> Celery worker process crashes

Why This Is Dangerous

This behavior introduces a serious stability and security risk:

• There is no supported extension point to intercept or validate the raw SQS message before JSON parsing.
• on_decode_error is never called, since the failure happens earlier in _message_to_python.
• Even returning None from _message_to_python causes additional crashes ('NoneType' object is not subscriptable), because higher layers assume the structure of a fully valid Celery message.
• A single malformed SQS message can kill all Celery consumers on that queue, causing a full outage.
• In real AWS environments, it is common to have:
  • accidentally published messages,
  • dev/test messages,
  • IAM misconfigurations,
  • manual AWS Console messages,
  • or third-party producers.

A production task processing system should not be this fragile.

What Is Needed

Celery/Kombu should provide a first-class, documented mechanism that allows developers to safely handle invalid raw messages, including:

1. A pre-decode validation hook (e.g. on_raw_message).
2. A hook specifically for JSON decode failures (e.g. on_raw_message_error).
3. An option to discard or log invalid messages without killing the entire worker.
4. The ability to mark such messages as handled (delete from SQS) or route them to a DLQ programmatically.
5. Avoid classifying malformed messages as "Unrecoverable" errors.

This would make Celery significantly more robust in distributed and cloud environments.

Why Existing Workarounds Are Not Enough

• DLQ does not help because the worker crashes before the message can be retried and moved.
• External sanitizer consumers defeat the purpose of Celery managing the queue and add operational overhead.
• Monkey-patching Kombu internals is brittle and breaks easily with updates.
• There is no clean and stable mechanism today to prevent Celery from crashing on invalid messages.

Proposed Solution

Introduce an official API layer between SQS receive and Celery decode, such as:

• before_message_decode(raw_message)
• on_invalid_raw_message(exception, raw_message)
• Configurable behavior, such as:
  • DROP_INVALID_RAW_MESSAGES = True
  • LOG_INVALID_MESSAGES = True

This would let Celery users:

• gracefully discard malformed messages,
• log or notify the event,
• keep the worker alive,
• maintain reliability in real-world environments.

Conclusion

A single malformed SQS message should not be able to bring down a Celery worker permanently.
Given that modern cloud systems frequently deal with mixed producers or accidental messages, this lack of raw-message error handling represents a real operational vulnerability.

Providing a safe and supported mechanism to intercept and handle invalid raw messages would dramatically increase Celery’s stability and adoption in cloud-native deployments.

Thank you for considering this improvement.

Related issues:

• JSON decode error "Expecting value: line 1 column 1 (char 0)" #1551
• SQS: JSONDecodeError when reading message from Amazon SQS predefined_queues #1181

[emrementese]

This issue has been open since 2020 as a “feature request”, but in practice it represents a significant stability and operational risk for any system running Celery with SQS. In real cloud environments, it is common for non-Celery or malformed messages to accidentally reach an SQS queue (misconfigured IAM roles, AWS Console testing, third-party integrations, etc.). When this happens, the worker crashes with a JSONDecodeError, leading to an avoidable outage.

This is not an isolated edge case—it affects all Celery/Kombu users relying on SQS. For that reason, it should not be treated as a lightweight enhancement request but rather as a core reliability and safety concern.

Revisiting this issue and providing an official mechanism—such as a safe on_invalid_message hook or a pre-decode validation path—would significantly improve Celery’s robustness in cloud-native environments. A single malformed message should never be able to terminate a worker process.

Addressing this would materially strengthen Celery’s fault-tolerance and developer experience.

Also

• https://github.com/celery/kombu/pull/1306/files

Despite the reference to PR #1306 as a related fix, that PR does not address the core problem described in this issue.

PR #1306 only improves handling of certain SQS/SNS edge cases (mainly around Base64-encoded bodies and non-Kombu JSON variations). It still assumes that the incoming message is ultimately valid JSON that can be passed through json.loads(...).

The issue here, however, is about arbitrary or malformed messages reaching an SQS queue — something that happens easily in real cloud environments (AWS Console tests, misconfigured producers, incorrect IAM permissions, third-party systems, etc.). In those cases:
• The message body is not valid JSON,
• json.loads(...) raises JSONDecodeError,
• The exception propagates,
• And the Celery worker process terminates.

PR #1306 does not introduce any mechanism to:
• validate raw SQS messages before JSON decoding,
• gracefully drop or ignore invalid payloads,
• catch JSONDecodeError at the transport level,
• or provide a hook for on_invalid_message.

Therefore, the core issue remains:
A single malformed message in SQS can still crash all Celery consumers.

This is fundamentally a stability and safety concern, not only an SNS/SQS decoding edge case. A proper solution requires a supported pathway to handle invalid raw messages without killing the worker process.

[emrementese]

update

I'm update kombu & celery latest version

Even though the JSONDecodeError crash is resolved in Kombu 5.6.1, invalid/non-Celery messages on SQS are still not actually deleted despite the worker logging “Received and deleted unknown message”; instead, the message remains in the queue and resurfaces after the visibility timeout. This indicates that _on_invalid_raw_message triggers the asynchronous delete_message path, but the async HTTP client silently fails or no-ops, resulting in the message looping indefinitely. This behavior makes SQS transport unreliable in real production setups, because a single malformed message (e.g., sent from AWS Console for testing) permanently pollutes the queue. It would be ideal for Kombu to guarantee actual deletion of such messages—either via a reliable sync boto3 fallback, or by exposing an official handler hook—because the current behavior still leaves queues in a stuck, unrecoverable state.

[auvipy]

would you mind coming with a improved solution as draft PR?

[emrementese]

@auvipy Of course. I'll see what I can do