Dead Letter Policy

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 the max_receive_count set on KubeMQ.QueuePolicy, automatically moves it to the dead-letter channel named by 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 nack_all, 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
• Elixir SDK installed ({:kubemq, "~> 1.0"} in mix.exs)

Code

channel = "elixir-queues.dead-letter-policy"
dlq = "elixir-queues.dead-letter-policy.dlq"
{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "elixir-qs-dlq-stream")

{:ok, handle} = KubeMQ.Client.queue_upstream(client)

msg = KubeMQ.QueueMessage.new(
  channel: channel,
  body: "Fragile message",
  policy: KubeMQ.QueuePolicy.new(
    max_receive_count: 2,
    max_receive_queue: dlq
  )
)

{:ok, _} = KubeMQ.QueueUpstreamHandle.send(handle, [msg])
IO.puts("Sent message with DLQ policy (max 2 receives)")

for attempt <- 1..2 do
  case KubeMQ.Client.poll_queue(client,
         channel: channel,
         max_items: 1,
         wait_timeout: 3_000
       ) do
    {:ok, poll} when length(poll.messages) > 0 ->
      IO.puts("Attempt #{attempt}: received, nacking...")
      {:ok, _} = KubeMQ.PollResponse.nack_all(poll)
      Process.sleep(500)

    _ ->
      IO.puts("Attempt #{attempt}: no message")
  end
end

Process.sleep(1_000)

case KubeMQ.Client.receive_queue_messages(client, dlq,
       max_messages: 10,
       wait_timeout: 3_000
     ) do
  {:ok, result} ->
    IO.puts("DLQ has #{result.messages_received} messages")
    Enum.each(result.messages, fn m -> IO.puts("  #{m.body}") end)

  {:error, _} ->
    IO.puts("DLQ empty (message may still be in transit)")
end

KubeMQ.QueueUpstreamHandle.close(handle)
KubeMQ.Client.close(client)

How It Works

• Messages are sent via the upstream stream with a QueuePolicy containing DLQ configuration
• Each nack increments the receive count; after max_receive_count, the message moves to the DLQ
• The DLQ channel can be monitored separately for failed message inspection

Related

• Dead Letter Queue
• Nack All