# Retries & Dead Letter Queues (/integrations/rayserve/how-to/dlq-and-retries)



ML inference handlers fail for two very different reasons. A transient fault — a model server that is briefly overloaded, a downstream API that times out — clears on a second attempt. A permanent fault — a malformed input, a bug in the handler, a missing dependency — never will. The KubeMQ Ray Serve adapter treats these distinctly: it retries a configurable number of times, then routes the exhausted task to a dead letter queue (DLQ) and fires a callback so you can alert on it. This guide covers retry configuration, the DLQ flow, the `on_dlq` hook, and the surrounding resilience behaviors that keep an expensive inference result from being lost to a transient broker blip.

## Prerequisites [#prerequisites]

* `kubemq-rayserve` installed and a `@task_consumer` deployment already running (see [Getting Started with Ray Serve](/integrations/rayserve/tutorials/getting-started))
* A running KubeMQ broker reachable from the adapter

## Configuring Retries and the DLQ [#configuring-retries-and-the-dlq]

Retry behavior is driven by two fields on the `TaskProcessorConfig` that Ray Serve passes to the adapter: `max_retries` and `failed_task_queue_name`. The adapter reads them in `initialize()` and applies them to every task it enqueues.

When both are set, `enqueue_task_sync` stamps each `QueueMessage` with KubeMQ's native redelivery controls — `max_receive_count` (how many times a message may be delivered before the broker gives up) and `max_receive_queue` (the channel the broker moves it to once that count is reached):

```python title="adapter.py (enqueue_task_sync)"
msg_kwargs: dict[str, Any] = {
    "channel": self._queue_name,
    "body": body,
}
if self._max_retries > 0 and self._failed_task_queue_name:
    msg_kwargs["max_receive_count"] = self._max_retries
    msg_kwargs["max_receive_queue"] = self._failed_task_queue_name
```

In a Ray Serve deployment you set these on the `TaskProcessorConfig` alongside the adapter class and config:

```python title="deployment.py"
from ray import serve
from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

@serve.deployment(ray_actor_options={"num_gpus": 1})
@task_consumer(
    task_processor_config=TaskProcessorConfig(
        adapter_class=KubeMQTaskProcessorAdapter,
        adapter_config=KubeMQAdapterConfig(address="kubemq:50000"),
        queue_name="inference-tasks",
        max_retries=3,
        failed_task_queue_name="inference-tasks-dlq",
    )
)
class ModelDeployment:
    def __init__(self):
        self.model = load_model()

    def __call__(self, data: str) -> dict:
        return self.model.predict(data)
```

For standalone usage outside a deployment, supply a small config object with the same fields:

```python title="standalone.py"
class TaskConfig:
    queue_name = "inference-tasks"
    max_retries = 3
    failed_task_queue_name = "inference-tasks-dlq"
    unprocessable_task_queue_name = ""

adapter.initialize(consumer_concurrency=2, task_processor_config=TaskConfig())
```

<Callout type="info">
  Both `max_retries > 0` and a non-empty `failed_task_queue_name` are required to activate redelivery. If either is missing, messages carry no retry controls and a handler exception simply nacks the task back to the source queue without a retry ceiling or DLQ destination.
</Callout>

## How a Retry Plays Out [#how-a-retry-plays-out]

The retry decision is made in the consumer's `_process_message` path. When a handler raises, the adapter inspects the message's `receive_count` and compares it against `max_retries`:

* If `receive_count < max_retries`, retries remain. The message is **nacked** so KubeMQ redelivers it to the same queue, and the handler runs again.
* Once retries are exhausted, the adapter stores a `FAILURE` result for the task, nacks the message (KubeMQ routes it to the `max_receive_queue` — the DLQ), increments `tasks_completed_total['FAILURE']`, and fires the `on_dlq` callback if one is configured.

```python title="adapter.py (_process_message — exception branch)"
except Exception as exc:
    receive_count = getattr(message, "receive_count", 0)
    if self._max_retries > 0 and receive_count < self._max_retries:
        # Retries remaining — nack for redelivery
        logger.warning(
            "Handler error (retries remaining=%d): task_id=%s error=%s",
            self._max_retries - receive_count, task_id, exc,
        )
        message.nack()
    else:
        # Retries exhausted — store FAILURE, nack to DLQ
        error_str = f"{type(exc).__name__}: {exc}"
        self._result_backend.store_result(
            task_id, "FAILURE", error=error_str, created_at=created_at
        )
        self._metrics.inc_completed("FAILURE")
        message.nack()

        # Fire DLQ callback
        if self._config.on_dlq is not None:
            try:
                self._config.on_dlq(task_id, error_str)
            except Exception as cb_exc:
                logger.error("on_dlq callback error: %s", cb_exc)
```

The full lifecycle of a failing task looks like this:

<Mermaid
  chart="flowchart TD
    Enqueue[&#x22;enqueue_task_sync()&#x22;] --> Queue[&#x22;KubeMQ Queue<br/>(inference-tasks)&#x22;]
    Queue --> Handler[&#x22;Handler runs&#x22;]
    Handler -->|&#x22;success&#x22;| Ack[&#x22;ack → SUCCESS&#x22;]
    Handler -->|&#x22;raises&#x22;| Check{&#x22;receive_count<br/>< max_retries?&#x22;}
    Check -->|&#x22;yes&#x22;| Nack[&#x22;nack → redeliver&#x22;]
    Nack --> Queue
    Check -->|&#x22;no (exhausted)&#x22;| DLQ[&#x22;store FAILURE<br/>nack → DLQ<br/>fire on_dlq&#x22;]"
/>

### Transient Failure: Retried to Success [#transient-failure-retried-to-success]

A handler that fails on its first attempt but succeeds on retry recovers automatically — no task is lost. The packaged `handler_exception` example demonstrates this with a flaky handler that fails once, then succeeds:

```python title="examples/error_handling/handler_exception.py"
call_count: dict[str, int] = {}

def flaky_handler(data: str) -> dict:
    """Handler that fails on first call but succeeds on retry."""
    count = call_count.get(data, 0) + 1
    call_count[data] = count
    print(f"  [handler] Attempt {count} for: {data}")

    if count < 2:
        raise RuntimeError(f"Transient error on attempt {count} for: {data}")

    return {"data": data, "succeeded_on_attempt": count}
```

With `max_retries=3`, the first invocation raises, the message is nacked and redelivered, and the second invocation returns the result — the task ends in `SUCCESS`.

### Permanent Failure: Exhausted to DLQ [#permanent-failure-exhausted-to-dlq]

A handler that always raises walks through every retry and lands in the DLQ. The `retries_exhausted_dlq` example pairs an always-failing handler with `max_retries=2`:

```python title="examples/error_handling/retries_exhausted_dlq.py"
def always_fails(data: str) -> dict:
    """Handler that always raises to demonstrate DLQ routing."""
    raise RuntimeError(f"Permanent failure processing: {data}")

class _Cfg:
    queue_name = channel
    max_retries = 2
    failed_task_queue_name = f"{channel}.dlq"
    unprocessable_task_queue_name = ""

adapter.initialize(consumer_concurrency=1, task_processor_config=_Cfg())
adapter.register_task_handler(always_fails, name="always_fails")
adapter.start_consumer()

result = adapter.enqueue_task_sync("always_fails", args=["doomed-work"])
# After 2 retries the task is routed to "{channel}.dlq" and its
# status becomes FAILURE.
```

Polling `get_task_status_sync(result.id)` eventually returns `FAILURE`, and the original message now sits in `{channel}.dlq` for inspection or reprocessing.

## Alerting with the on\_dlq Callback [#alerting-with-the-on_dlq-callback]

Storing a `FAILURE` result is passive — something has to look at it. The `on_dlq` callback turns DLQ routing into an active signal. It is configured on `KubeMQAdapterConfig` and has the signature `(task_id: str, error: str) -> None`:

```python title="config.py"
on_dlq: Callable[[str, str], None] | None = Field(default=None, exclude=True)
```

The callback fires **only after retries are exhausted** — once per permanently failed task, not on every retry. Wire it to whatever alerting surface you use (PagerDuty, Slack, a metrics counter, structured logging):

```python title="examples/error_handling/dlq_monitoring_callback.py"
dlq_events: list[dict] = []

def on_dlq_callback(task_id: str, error: str) -> None:
    """Called when a task exhausts retries and moves to DLQ."""
    print(f"  [DLQ ALERT] task_id={task_id} error={error}")
    dlq_events.append({"task_id": task_id, "error": error})

config = KubeMQAdapterConfig(
    address="localhost:50000",
    on_dlq=on_dlq_callback,
)
```

<Callout type="info">
  Exceptions raised inside your callback are caught and logged by the adapter — a buggy alerting hook will not crash the consumer or break task processing. The callback runs inline on the consumer thread, so keep it fast: hand off slow work (network calls, paging) to a queue or background task rather than blocking the loop.
</Callout>

## Malformed Messages [#malformed-messages]

A handler exception is a *runtime* failure of valid work. A malformed message is different: the body cannot be deserialized at all, so there is no handler to run and no point retrying. The adapter handles this case separately, before dispatch. When `deserialize_task_payload` raises `JSONDecodeError` or `UnicodeDecodeError`, the message is **acked** (removed from the queue so it never blocks the consumer) and re-sent to a dedicated channel — `unprocessable_task_queue_name` — which is distinct from the retry DLQ:

```python title="adapter.py (_process_message — deserialize branch)"
except (json.JSONDecodeError, UnicodeDecodeError) as exc:
    logger.warning("Malformed message (acking to unprocessable): %s", exc)
    message.ack()
    # Move to unprocessable queue via re-queue if configured
    if self._unprocessable_task_queue_name:
        try:
            msg = QueueMessage(
                channel=self._unprocessable_task_queue_name,
                body=message.body,
            )
            self._client.send_queue_message(msg)
        except Exception as send_exc:
            logger.error("Failed to send to unprocessable queue: %s", send_exc)
    return
```

The `malformed_message` example sends raw non-JSON bytes directly through the SDK, then enqueues a valid task and confirms it processes normally — proving a bad message does not stall the queue:

```python title="examples/error_handling/malformed_message.py"
class _Cfg:
    queue_name = channel
    max_retries = 0
    failed_task_queue_name = ""
    unprocessable_task_queue_name = f"{channel}.unprocessable"

# ... a raw, non-JSON body is sent straight to the channel:
malformed_body = b"this is not valid JSON {{{{"
raw_client.send_queue_message(QueueMessage(channel=channel, body=malformed_body))
```

<Callout type="info">
  Always set `unprocessable_task_queue_name` in production. Without it, a malformed message is still acked and dropped from the source queue (so it cannot block processing), but its body is discarded rather than preserved for later inspection. The two destinations serve different purposes: the DLQ (`failed_task_queue_name`) holds *valid* tasks that failed processing; the unprocessable queue holds *undecodable* bytes.
</Callout>

## Serialization Errors at Enqueue Time [#serialization-errors-at-enqueue-time]

Some failures are caught even earlier — before a message ever reaches the broker. Task arguments are serialized to JSON in `serialize_task_payload`. If an argument is not JSON-serializable (a `set`, a NumPy `ndarray`, a custom object), `json.dumps` raises `TypeError`, and the adapter re-raises it with actionable guidance:

```python title="_internal/serialization.py"
try:
    return json.dumps(payload).encode("utf-8")
except TypeError as exc:
    raise TypeError(
        f"Task argument serialization failed: {exc}. "
        "Convert complex objects before enqueueing (e.g., ndarray.tolist()). "
        "Pluggable serializers planned for v1.1."
    ) from exc
```

This surfaces synchronously at the `enqueue_task_sync` call, so you find the problem in the producer rather than discovering a stuck task later. The fix is to convert objects to JSON-native types first:

```python title="examples/error_handling/serialization_error.py"
# This raises TypeError: set() is not JSON-serializable
adapter.enqueue_task_sync("process", kwargs={"data": {1, 2, 3}})

# Fixed: convert the set to a list before enqueueing
result = adapter.enqueue_task_sync("process", kwargs={"data": [1, 2, 3]})
```

## Result-Storage Resilience [#result-storage-resilience]

Retries and DLQs protect the *work*. The result backend protects the *output*. ML inference is expensive — losing a completed result to a transient broker hiccup would mean re-running the whole computation. To avoid that, `store_result` retries the write up to three times with a one-second backoff, then falls back to in-memory storage, and **never re-raises**:

```python title="result_backend.py (store_result — retry loop)"
for attempt in range(1, _STORE_MAX_RETRIES + 1):
    try:
        msg = QueueMessage(
            channel=channel, body=body, expiration_in_seconds=self._expiry,
        )
        self._client.send_queue_message(msg)
        return
    except Exception as exc:
        self._retries_total += 1
        if attempt < _STORE_MAX_RETRIES:
            logger.warning(
                "Result storage retry attempt=%d task_id=%s error=%s",
                attempt, task_id, exc,
            )
            time.sleep(_STORE_RETRY_BACKOFF_SECONDS)
        else:
            logger.error(
                "Result storage failed after %d retries task_id=%s",
                _STORE_MAX_RETRIES, task_id,
            )
            # Fallback to in-memory (with eviction guard)
            self._fallback[task_id] = {
                "task_id": task_id, "status": "FAILURE",
                "result": None, "error": f"Result storage failed: {exc}",
            }
```

The retry constants are `_STORE_MAX_RETRIES = 3` and `_STORE_RETRY_BACKOFF_SECONDS = 1.0`. A common trigger is an oversized result that exceeds `max_send_size` (default 4 MB). The `result_storage_failure` example shows the graceful-degradation outcome: a 5 MB result cannot be transmitted, so the backend logs a warning and the task remains `SUCCESS` with a `None` result — the handler's work is never counted as a failure just because the payload was too large to store.

```python title="examples/error_handling/result_storage_failure.py"
def large_result_handler(data: str) -> dict:
    """Returns a very large result that may exceed max_send_size."""
    large_payload = "x" * (5 * 1024 * 1024)  # ~5MB
    return {"data": data, "payload": large_payload}
```

<Callout type="warn">
  For genuinely large outputs (image batches, embeddings, model artifacts), write them to external storage — object store, blob, shared volume — and return a *reference* (a URL or key) as the task result instead of the bytes themselves.
</Callout>

## Connection and Shutdown Handling [#connection-and-shutdown-handling]

The final layer of resilience covers the broker connection itself and clean shutdown.

**Connection errors.** The consumer loop catches `KubeMQConnectionError` (and `KubeMQConnectionNotReadyError`), logs a warning, sleeps briefly, and keeps looping — it does not exit. The underlying reconnection is handled by the KubeMQ SDK, so no adapter-level reconnect logic is needed:

```python title="adapter.py (_consumer_loop)"
except (KubeMQConnectionError, KubeMQConnectionNotReadyError) as exc:
    logger.warning("Consumer connection error (will retry): %s", exc)
    if self._running:
        time.sleep(1)
```

If you stop the broker mid-run, you may see gRPC errors in the logs during the disconnect window; once it returns, the SDK reconnects transparently and the consumer resumes. The `connection_error` and `reconnection_behavior` examples exercise an unreachable broker and a manual restart cycle respectively.

**Graceful shutdown.** `stop_consumer()` cancels the query-subscription token, sets `_running = False` to stop the polling loop, joins the consumer thread, and closes the SDK clients. Messages still in flight when shutdown begins are nacked inside the consumer loop, so they return to the queue for redelivery rather than being silently lost:

```python title="adapter.py (_consumer_loop — shutdown nack)"
for message in response.messages:
    if not self._running:
        # Shutting down — nack remaining messages
        message.nack()
        continue
    self._process_message(message)
```

The `graceful_shutdown` example enqueues a five-second handler, then calls `stop_consumer(timeout=10.0)` while the task is in flight, and inspects the final status afterward:

```python title="examples/error_handling/graceful_shutdown.py"
adapter.stop_consumer(timeout=10.0)
print("  Consumer stopped.")
# An in-flight task is nacked for re-delivery, or completes before
# shutdown finishes — either way it is not lost.
```

## Related [#related]

* [API Reference](/integrations/rayserve/reference/api) — adapter methods and the metrics dict (including `dlq_depth`, which reports the live DLQ message count).
* [Configuration Reference](/integrations/rayserve/reference/configuration) — every `KubeMQAdapterConfig` field, including `on_dlq`.
* [GPU Multi-Model Inference Service](/integrations/rayserve/scenarios/gpu-multi-model-service) — applies these retry and DLQ patterns to GPU-backed model deployments on Kubernetes.