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:

docker run -d \  --name kubemq \  -p 50000:50000 \  -p 9090:9090 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

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

kubemq-celery installed and a Celery app already pointed at a kubemq:// broker URL (see Configuration)
The broker above running and reachable

Task Retry with self.retry()

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.

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:

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

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:

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:

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

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:

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

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:

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

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.

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.

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:

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:

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

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:

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

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:

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.

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

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:

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:

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

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:

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 captures task failures with full tracebacks, retries, performance spans, and distributed traces across chains, chords, and groups — automatically, once you register CeleryIntegration:

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:

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

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:

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:

The worker detects connection loss (a keepalive timeout or a send/receive error).
A KubeMQConnectionError is raised and classified as a recoverable connection error.
Kombu's connection recovery kicks in.
A new gRPC channel is established to the broker.
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.

Error Handling & Dead Letter Queues