# Error Handling & Dead Letter Queues (/integrations/celery/how-to/error-handling)
Resilient task processing combines two independent layers: Celery's application-level retry logic (`self.retry()`, `max_retries`, backoff) and KubeMQ's broker-level delivery guarantees (native ack/nack, `delay_in_seconds`, and a dead letter queue). This guide shows how to use both together — distinguishing transient from permanent failures, capping poison messages with a DLQ, choosing the right acknowledgment mode, designing idempotent tasks for redelivery, and wiring failures into Sentry.
All examples assume a running KubeMQ broker. The fastest way to get one is Docker:
Port `50000` is the gRPC port the transport connects to. Port `9090` is the shared HTTP server (REST and the `/health` probe) — `curl http://localhost:9090/health` confirms the broker is up as tasks flow through.
## Prerequisites [#prerequisites]
* `kubemq-celery` installed and a Celery app already pointed at a `kubemq://` broker URL (see [Configuration](/integrations/celery/how-to/configuration))
* The broker above running and reachable
## Task Retry with self.retry() [#task-retry-with-selfretry]
Bind a task with `bind=True` to get access to `self`, then call `self.retry()` on transient failures. The retry re-dispatches a new message through KubeMQ with an optional `countdown` delay — implemented with KubeMQ's native server-side `delay_in_seconds`, so there is no client-side polling and no worker memory held while the message waits.
The key discipline is distinguishing **transient** errors (network blips, rate limits, temporary timeouts) that are worth retrying from **permanent** errors (validation failures, bad input) that will never succeed and should fail immediately.
```python title="basic_retry.py"
import kubemq_celery # registers the kubemq:// transport
from celery import Celery
app = Celery("myapp", broker="kubemq://localhost:50000")
@app.task(bind=True, max_retries=3)
def send_email(self, to: str, subject: str, body: str) -> dict:
"""Send email with automatic retry on transient failures."""
try:
result = email_service.send(to=to, subject=subject, body=body)
return {"status": "sent", "message_id": result.id}
except ConnectionError as exc:
# Transient: retry with exponential backoff -- 10s, 20s, 40s
raise self.retry(exc=exc, countdown=10 * (2 ** self.request.retries))
except ValueError as exc:
# Permanent error -- do NOT retry
logger.error("Invalid email params: %s", exc)
raise
```
* `bind=True` gives access to `self` for retry control and `self.request.retries` (the current attempt count).
* `max_retries=3` caps the number of retry attempts.
* `countdown` maps to KubeMQ's native server-side delay — no client-side polling.
* Catch only the exceptions you know are transient; let permanent errors propagate so the task fails fast.
If you would rather let Celery retry automatically without an explicit `try/except`, use `autoretry_for` and `default_retry_delay`:
```python title="auto_retry.py"
@app.task(
autoretry_for=(ConnectionError, TimeoutError),
max_retries=3,
default_retry_delay=1,
)
def auto_retry_task(url: str) -> dict:
"""Celery automatically retries on the listed exceptions."""
response = requests.get(url, timeout=10)
response.raise_for_status()
return {"url": url, "status": "fetched"}
```
## Exponential Backoff with Jitter [#exponential-backoff-with-jitter]
For external API calls, back off exponentially and add jitter so a fleet of workers does not retry in lockstep (the "thundering herd"). You can compute the schedule manually:
```python title="exponential_backoff.py"
import random
@app.task(bind=True, max_retries=5, default_retry_delay=60)
def call_external_api(self, endpoint: str, payload: dict) -> dict:
"""Call an external API with capped exponential backoff and jitter."""
try:
response = requests.post(endpoint, json=payload, timeout=10)
response.raise_for_status()
return response.json()
except requests.RequestException as exc:
backoff = min(60 * (2 ** self.request.retries), 600) # cap at 10 min
jitter = random.uniform(0, backoff * 0.1) # 10% jitter
raise self.retry(exc=exc, countdown=backoff + jitter)
```
This produces a backoff schedule of **60s → 120s → 240s → 480s → 600s** (capped at 10 minutes), with up to 10% random jitter added to each step.
Or let Celery do the backoff math for you with `retry_backoff`, `retry_backoff_max`, and `retry_jitter`:
```python title="custom_retry_policy.py"
@app.task(
autoretry_for=(ConnectionError, TimeoutError),
max_retries=5,
retry_backoff=True, # exponential backoff: 1, 2, 4, 8, 16...
retry_backoff_max=60, # cap delay at 60 seconds
retry_jitter=True, # randomize to prevent thundering herd
)
def resilient_http_call(url: str) -> dict:
response = requests.get(url, timeout=10)
response.raise_for_status()
return {"url": url, "status": "ok"}
```
KubeMQ's maximum server-side delay is **86400 seconds (24 hours)**. Countdown values that exceed this are capped automatically, so a runaway backoff schedule cannot push a task arbitrarily far into the future.
## Dead Letter Queue [#dead-letter-queue]
Celery's `max_retries` governs application-level retries. KubeMQ's dead letter queue operates one level lower — at the **broker receive-count level** — and catches "poison" messages that fail repeatedly regardless of task logic (for example, a message a worker keeps crashing on before it can ack). Configure it through `broker_transport_options`:
```python title="dead_letter_queue.py"
import os
import kubemq_celery # registers the kubemq:// transport
from celery import Celery
app = Celery("dead_letter_queue")
app.config_from_object(
{
"broker_url": os.environ.get("CELERY_BROKER_URL", "kubemq://localhost:50000"),
"broker_transport_options": {
# After 5 failed receive attempts, the message moves to the DLQ.
"dead_letter_queue": "celery-dead-letters",
"max_receive_count": 5,
},
}
)
```
| Option | Type | Default | Description |
| ------------------- | ----- | ------- | ----------------------------------------------------------------------------------------------------------------- |
| `dead_letter_queue` | `str` | `""` | KubeMQ channel name for dead letter messages. Messages that exceed `max_receive_count` are routed here. |
| `max_receive_count` | `int` | `0` | Maximum receive attempts before routing to the DLQ. `0` disables the DLQ (messages are redelivered indefinitely). |
Setting `max_receive_count > 0` **requires** `dead_letter_queue` to be set — otherwise the transport raises `KubeMQCeleryConfigError` at startup.
The DLQ is independent of Celery's retry mechanism — the two count different things and can be used together. `max_retries` counts how many times your task code asked to retry; `max_receive_count` counts how many times the broker delivered the message before it was acknowledged. A message that exhausts `max_receive_count` is moved off the working queue so healthy tasks keep flowing.
### Inspecting DLQ Messages [#inspecting-dlq-messages]
Because the DLQ is just another KubeMQ Queue channel, you can inspect it two ways.
Query queue statistics over the shared HTTP server on `9090` with the REST `/queue/info` endpoint — `curl http://localhost:9090/queue/info`. This is the quickest way to confirm messages are landing in the `celery-dead-letters` channel during development. For a full dashboard view, open the KubeMQ Management API dashboard on `:8080`.
Read DLQ messages from your own code with a non-destructive `peek_queue_messages()` call — peeking does not consume the message, so you can inspect without removing it from the queue:
```python title="inspect_dlq.py"
from kubemq.queues.client import Client as QueuesClient
from kubemq.core import ClientConfig
client = QueuesClient(config=ClientConfig(address="localhost:50000"))
response = client.peek_queue_messages(
channel="celery-dead-letters",
max_messages=10,
wait_timeout_in_seconds=1,
)
for msg in response.messages:
print(f"DLQ message: {msg.body.decode()}")
```
## acks\_early vs acks\_late [#acks_early-vs-acks_late]
The acknowledgment mode decides *when* a message is acked relative to task execution, which in turn decides what happens if a worker dies mid-task.
The message is acknowledged **before** the task executes. If the worker crashes during execution, the task is **not** redelivered — you rely on `self.retry()` for application-level recovery. This is the safest and most performant option for most tasks: with `task_acks_late=False`, the transport receives with `auto_ack=True`, and KubeMQ's native ack eliminates the visibility-timeout race that causes duplicate execution on Redis.
```python title="acks_early.py"
app.conf.task_acks_late = False # default
```
The message is acknowledged **after** the task completes successfully (at-least-once delivery). If the worker crashes, KubeMQ redelivers the message to another worker. Pair it with `task_reject_on_worker_lost=True` so a killed worker (OOM, `SIGKILL`) nacks the message — mapping to KubeMQ's native `nack()` — and a `worker_prefetch_multiplier` of `1` so a worker only holds one unacked message at a time.
```python title="acks_late.py"
import os
import kubemq_celery
from celery import Celery
app = Celery("task_acks_late")
app.config_from_object(
{
"broker_url": os.environ.get("CELERY_BROKER_URL", "kubemq://localhost:50000"),
"task_acks_late": True,
"task_reject_on_worker_lost": True, # nack on worker crash
"worker_prefetch_multiplier": 1,
}
)
@app.task
def critical_operation(item_id: str) -> dict:
"""A task where message loss is unacceptable.
1. Worker receives the message (not acked yet)
2. Task executes
3. Success -> ack (message consumed)
4. Worker crash -> nack (message redelivered by KubeMQ)
"""
return {"item_id": item_id, "status": "completed"}
```
You can also override the mode per task instead of globally:
```python title="per_task_override.py"
@app.task(acks_late=True, reject_on_worker_lost=True)
def critical_payment(payment_id: str) -> dict:
"""Process a payment -- must not be lost. Ensure idempotency via payment_id."""
...
```
`acks_late` interacts with KubeMQ's transaction model. For **short tasks (\< \~60s)** it works as expected. For **long-running tasks (> \~60s)** the KubeMQ server-side transaction timeout can expire before the task finishes, causing the message to be redelivered and the task to run twice. For long tasks, either use `acks_early` (the default) or make the task idempotent and accept at-least-once semantics.
For finer control, raise `Reject` to nack a message explicitly — requeue it for a later attempt, or discard it permanently:
```python title="reject_and_requeue.py"
from celery.exceptions import Reject
@app.task(bind=True, max_retries=0)
def process_if_ready(self, job_id: str) -> dict:
if not external_system_ready():
# Requeue -- calls KubeMQ's native re_queue() on the message
raise Reject(reason="external system not ready", requeue=True)
return {"job_id": job_id, "status": "processed"}
@app.task(bind=True, max_retries=0)
def validate_and_process(self, data: dict) -> dict:
if not data.get("required_field"):
# Discard permanently -- no requeue
raise Reject(reason="missing required_field", requeue=False)
return {"data": data, "status": "valid_and_processed"}
```
## Handling MaxRetriesExceededError [#handling-maxretriesexceedederror]
When every retry attempt is exhausted, `self.retry()` raises `MaxRetriesExceededError`. Catch it to escalate gracefully — log, notify, or trigger compensating logic — then re-raise so Celery still marks the task as `FAILURE`:
```python title="max_retries_exceeded.py"
from celery.exceptions import MaxRetriesExceededError
@app.task(bind=True, max_retries=3)
def process_order(self, order_id: str) -> dict:
try:
return do_process(order_id)
except TransientError as exc:
try:
raise self.retry(exc=exc, countdown=30)
except MaxRetriesExceededError:
# All retries exhausted -- log and escalate
logger.error("Order %s failed after %d retries", order_id, self.max_retries)
notify_ops_team(order_id=order_id, error=str(exc))
raise # let Celery mark the task as FAILURE
```
## Idempotent Task Design [#idempotent-task-design]
Any time a task can run more than once — `acks_late` redelivery, broker-level redelivery, or an aggressive retry policy — it **must** be idempotent: running it twice with the same input must produce the same result and no duplicate side effects.
The most common pattern is a cache- or DB-backed dedup check keyed on a stable identifier:
```python title="idempotent_cache.py"
@app.task(bind=True, acks_late=True)
def process_upload(self, file_id: str, checksum: str) -> dict:
"""Idempotent file processing -- safe for re-delivery.
Uses the file checksum as the idempotency key. If already
processed, returns the cached result without re-processing.
"""
cache_key = f"processed:{checksum}"
cached = cache.get(cache_key)
if cached:
logger.info("File %s already processed (idempotent skip)", file_id)
return cached
result = do_expensive_processing(file_id)
cache.set(cache_key, result, timeout=86400)
return result
```
A complementary approach is a **deterministic task ID**: derive the `task_id` from the task name and arguments so the same call always produces the same ID. You can then check for an existing result before dispatching a duplicate.
```python title="idempotent_tasks.py"
import hashlib
from celery import Task
from celery.result import AsyncResult
def deterministic_task_id(task_name: str, *args, **kwargs) -> str:
"""Same arguments always produce the same ID, enabling deduplication."""
key = f"{task_name}:{args}:{sorted(kwargs.items())}"
return hashlib.sha256(key.encode()).hexdigest()[:32]
def dispatch_idempotent(task: Task, *args, **kwargs) -> AsyncResult:
task_id = deterministic_task_id(task.name, *args, **kwargs)
existing = AsyncResult(task_id, app=app)
if existing.state not in ("PENDING",):
return existing # dedup hit -- already dispatched
return task.apply_async(args=args, kwargs=kwargs, task_id=task_id)
```
Other strategies that pair well with these:
* **Database constraints** — a unique constraint rejects duplicate writes at the storage layer.
* **Conditional updates** — `UPDATE ... WHERE version = N` makes a write a no-op on the second attempt.
## Error Callbacks and on\_failure Hooks [#error-callbacks-and-on_failure-hooks]
For per-dispatch cleanup or notification, attach error callbacks with `link_error`. Each callback receives the failed task's ID (as a string) and runs only when the linked task fails:
```python title="error_callbacks.py"
@app.task
def notify_failure(task_id: str) -> dict:
"""link_error callback -- receives the failed task's ID."""
alert_ops_team(task_id)
return {"task_id": task_id, "notification": "sent"}
@app.task
def rollback_payment(task_id: str) -> dict:
rollback(task_id)
return {"task_id": task_id, "rollback": "completed"}
# Dispatch with one or more error callbacks
process_payment.apply_async(
args=("ORD-002", 15000.00),
link_error=[notify_failure.s(), rollback_payment.s()],
)
```
For per-task error logic that travels with the task definition, override `on_failure()` in a custom `Task` base class:
```python title="on_failure_callback.py"
from billiard.einfo import ExceptionInfo
class AlertingTask(app.Task):
"""Custom base task that hooks into the failure lifecycle."""
def on_failure(self, exc, task_id, args, kwargs, einfo: ExceptionInfo) -> None:
logger.error("Task %s failed: %s (args=%s)", task_id, exc, args)
@app.task(base=AlertingTask, bind=True, max_retries=0)
def risky_operation(self, item_id: str) -> dict:
raise RuntimeError(f"Cannot process item {item_id}")
```
## Monitoring Failures with Signals and Sentry [#monitoring-failures-with-signals-and-sentry]
For cross-cutting monitoring that spans every task, connect Celery's `task_failure` and `task_retry` signals. These fire regardless of which task failed, so they are the right place for centralized logging and metrics:
```python title="signals.py"
from celery import signals
@signals.task_failure.connect
def on_task_failure(sender=None, task_id=None, exception=None, traceback=None, **kwargs):
logger.error(
"Task %s[%s] failed: %s",
sender.name if sender else "unknown",
task_id,
exception,
)
@signals.task_retry.connect
def on_task_retry(sender=None, request=None, reason=None, **kwargs):
logger.warning(
"Task %s[%s] retrying: %s (attempt %d)",
sender.name if sender else "unknown",
request.id if request else "?",
reason,
request.retries if request else 0,
)
```
These signals work with any backend — Prometheus, Datadog, or plain logging.
### Sentry Integration [#sentry-integration]
[Sentry](https://sentry.io/) captures task failures with full tracebacks, retries, performance spans, and distributed traces across chains, chords, and groups — automatically, once you register `CeleryIntegration`:
```python title="sentry_setup.py"
import os
import sentry_sdk
from sentry_sdk.integrations.celery import CeleryIntegration
sentry_sdk.init(
dsn=os.environ["SENTRY_DSN"],
integrations=[CeleryIntegration()],
traces_sample_rate=0.1, # 10% of transactions
profiles_sample_rate=0.1,
environment=os.environ.get("ENVIRONMENT", "development"),
release=os.environ.get("APP_VERSION", "unknown"),
)
```
Retries are expected behavior, so filter `MaxRetriesExceededError` out with `before_send` to keep Sentry signal high while still capturing unexpected errors:
```python title="sentry_filter.py"
sentry_sdk.init(
dsn=os.environ["SENTRY_DSN"],
integrations=[CeleryIntegration()],
before_send=lambda event, hint: (
None if "MaxRetriesExceededError" in str(hint.get("exc_info", ""))
else event
),
)
```
## Recovering from Dropped Connections [#recovering-from-dropped-connections]
Networks fail, brokers restart, and load balancers time out idle connections. The transport recovers from this with two cooperating layers: **gRPC keepalive** detects a stale connection quickly, and **Celery's `broker_connection_retry`** re-establishes it. Configure both in your app:
```python title="reconnection.py"
import os
import kubemq_celery
from celery import Celery
app = Celery("reconnection")
app.config_from_object(
{
"broker_url": os.environ.get("CELERY_BROKER_URL", "kubemq://localhost:50000"),
# Retry connecting to the broker on worker startup
"broker_connection_retry_on_startup": True,
"broker_connection_max_retries": 10,
"broker_connection_retry": True,
"broker_transport_options": {
"connection_timeout": 10.0,
# Keepalive detects broken connections faster
"grpc_keepalive_time": 15, # ping every 15s
"grpc_keepalive_timeout": 5, # wait 5s for a response
},
}
)
```
The recovery flow:
1. The worker detects connection loss (a keepalive timeout or a send/receive error).
2. A `KubeMQConnectionError` is raised and classified as a recoverable connection error.
3. Kombu's connection recovery kicks in.
4. A new gRPC channel is established to the broker.
5. The worker resumes consuming tasks.
To verify, start a worker, restart the KubeMQ broker while it is running, and watch the worker auto-reconnect and resume processing without manual intervention.
## Related [#related]