Retries & Dead Letter Queues
Configure retry limits, route failures to a DLQ, and alert on permanent failures with the on_dlq callback.
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
kubemq-rayserveinstalled and a@task_consumerdeployment already running (see Getting Started with Ray Serve)- A running KubeMQ broker reachable from the adapter
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):
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_nameIn a Ray Serve deployment you set these on the TaskProcessorConfig alongside the adapter class and config:
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:
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())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.
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
FAILUREresult for the task, nacks the message (KubeMQ routes it to themax_receive_queue— the DLQ), incrementstasks_completed_total['FAILURE'], and fires theon_dlqcallback if one is configured.
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:
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:
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
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:
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
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:
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):
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,
)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.
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:
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)
returnThe 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:
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))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.
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:
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 excThis 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:
# 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
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:
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.
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}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.
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:
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:
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:
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
- API Reference — adapter methods and the metrics dict (including
dlq_depth, which reports the live DLQ message count). - Configuration Reference — every
KubeMQAdapterConfigfield, includingon_dlq. - GPU Multi-Model Inference Service — applies these retry and DLQ patterns to GPU-backed model deployments on Kubernetes.
Was this page helpful?