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 QueueMessagePolicy, 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
• Ruby SDK installed (gem install kubemq)

Code

require 'kubemq'

address = ENV.fetch('KUBEMQ_ADDRESS', 'localhost:50000')
channel = 'ruby-queues.dead-letter-policy'
dlq_channel = 'ruby-queues.dead-letter-policy-dlq'

begin
  client = KubeMQ::QueuesClient.new(address: address, client_id: 'qstream-dlq-example')
  puts "Connected to #{address}"

  policy = KubeMQ::Queues::QueueMessagePolicy.new(max_receive_count: 2, max_receive_queue: dlq_channel)
  msg = KubeMQ::Queues::QueueMessage.new(
    channel: channel,
    metadata: 'fragile-message',
    body: 'may-fail',
    policy: policy
  )
  client.send_queue_message(msg)
  puts "Sent message with max_receive_count=2, DLQ=#{dlq_channel}"

  receiver = client.create_downstream_receiver

  2.times do |attempt|
    request = KubeMQ::Queues::QueuePollRequest.new(channel: channel, max_items: 1, wait_timeout: 3)
    response = receiver.poll(request)
    puts "Attempt #{attempt + 1}: #{response.messages.size} messages"
    response.nack_all if response.messages.any?
  end

  sleep 1
  dlq_request = KubeMQ::Queues::QueuePollRequest.new(channel: dlq_channel, max_items: 1, wait_timeout: 3)
  dlq_response = receiver.poll(dlq_request)
  puts "DLQ messages: #{dlq_response.messages.size}"
  dlq_response.messages.each do |m|
    puts "  DLQ: #{m.metadata}"
    m.ack
  end
rescue KubeMQ::Error => e
  puts "KubeMQ error: #{e.message}"
ensure
  receiver&.close
  client&.close
  puts 'Done'
end

How It Works

• max_receive_count limits delivery attempts before the broker routes the message to max_receive_queue.
• This provides automatic poison-message handling without application-level retry logic.
• Review timeouts, channel names, and client IDs before running against shared environments.

Related

• Pattern overview
• Ruby SDK Reference
• Dead Letter Queue
• Delay Policy