KubeMQ
Client SDKsPythonHow-to guidesQueues

Dead Letter Policy

Configure a dead-letter policy on KubeMQ stream queues with the Python SDK to handle failed messages.

Which to use

This page is the field-level reference for max_receive_count/max_receive_queue, shown here via the queue-stream API. For the end-to-end task — sending a message, exhausting retries, and consuming from the resulting DLQ — see Dead Letter Queue.

Overview

A dead-letter policy protects a queue from poison messages — a record that fails processing over and over because of a malformed payload, a consumer bug, or a downstream dependency that is down. Without one, that message is redelivered forever: it blocks head-of-line delivery, burns your consumers' retry budget, and can stall an entire queue behind a single bad record.

With a policy attached, KubeMQ counts each failed delivery and, once the message crosses max_receive_count, automatically moves it to the dead-letter channel you name with max_receive_queue. The main queue keeps flowing while the failure is quarantined for inspection or replay.

Gotchas: the receive count increments on every failed delivery — an explicit async_nack(), an expired transaction, or a visibility timeout — not just deliberate rejections, so set the ceiling above your normal retry budget. The dead-letter channel is an ordinary queue with no special behavior: nothing drains it for you, so monitor it and build a reprocessing path or failures pile up silently. The policy is set at send time and travels with the message, so the producer, not the consumer, decides the retry ceiling.

Prerequisites

  • KubeMQ server running on localhost:50000
  • Python SDK installed (pip install kubemq)

Code

dead_letter_policy.py
"""Example: Dead letter policy — messages move to DLQ after max attempts via stream."""

from __future__ import annotations

import asyncio

from kubemq import AsyncQueuesClient
from kubemq import QueueMessage


async def main() -> None:
    async with AsyncQueuesClient(
        address="localhost:50000",
        client_id="python-queues-stream-dead-letter-policy-client",
    ) as client:
        # Send a message with combined policies: expiration + DLQ
        result = await client.send_queue_message(
            QueueMessage(
                channel="python-queues-stream.dead-letter-policy",
                body=b"message with policies",
                metadata="policy-test",
                expiration_in_seconds=60,
                delay_in_seconds=0,
                max_receive_count=3,
                max_receive_queue="python-queues-stream.dead-letter-policy-dlq",
            )
        )
        print(f"Sent with policies: {result}")
        print("  Expiration: 60s, Max attempts: 3, DLQ: dead-letter-policy-dlq")

        # Reject the message repeatedly to trigger DLQ
        for attempt in range(3):
            response = await client.receive_queue_messages(
                channel="python-queues-stream.dead-letter-policy",
                max_messages=1,
                wait_timeout_seconds=5,
            )
            if not response.messages:
                print(f"  Attempt {attempt + 1}: No message available")
                break
            for msg in response.messages:
                print(
                    f"  Attempt {attempt + 1}: receive_count={msg.receive_count}, rejecting..."
                )
                await msg.async_nack()

        # Check the DLQ
        dlq = await client.receive_queue_messages(
            channel="python-queues-stream.dead-letter-policy-dlq",
            max_messages=1,
            wait_timeout_seconds=5,
            auto_ack=True,
        )
        if dlq.messages:
            print(f"DLQ received: {dlq.messages[0].body.decode('utf-8')}")
        else:
            print("DLQ: no messages yet")


if __name__ == "__main__":
    asyncio.run(main())

How It Works

  • The message is sent with three combined policies: expiration_in_seconds=60 (TTL), max_receive_count=3 (max attempts), and max_receive_queue (DLQ target).
  • Each async_nack() increments receive_count; after 3 rejections the broker routes the message to the DLQ channel instead of returning it.
  • expiration_in_seconds is an independent TTL; if the message is not delivered within 60 seconds it expires whether or not it has been received.
  • The DLQ channel is a standard queue; a separate consumer can drain it for debugging, alerting, or reprocessing workflows.

Was this page helpful?

Batch Send

Send multiple KubeMQ queue messages in a single batch using the Python SDK for higher throughput.

Dead Letter Queue

Route failed KubeMQ queue messages to a dead-letter queue using the Python SDK for safe handling.

On this page

OverviewPrerequisitesCodeHow It WorksRelated