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 SetMaxReceiveCount, automatically moves it to the dead-letter channel you name with SetMaxReceiveQueue. 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, 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
• Go SDK installed (go get github.com/kubemq-io/kubemq-go/v2)

Code

// Example: queues-stream/dead-letter-policy
//
// Demonstrates sending queue messages with a dead-letter queue policy
// via the upstream stream. After exceeding the max receive count,
// messages are moved to the specified dead-letter queue.
//
// Channel: go-queues-stream.dead-letter-policy
// Client ID: go-queues-stream-dead-letter-policy-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).
package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/kubemq-io/kubemq-go/v2"
)

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	client, err := kubemq.NewClient(ctx,
		kubemq.WithAddress("localhost", 50000),
		kubemq.WithClientId("go-queues-stream-dead-letter-policy-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-queues-stream.dead-letter-policy"
	dlqChannel := "go-queues-stream.dead-letter-policy.dlq"

	// Send a message with dead-letter queue policy via upstream stream.
	upstream, err := client.QueueUpstream(ctx)
	if err != nil {
		log.Fatal(err)
	}
	defer upstream.Close()

	dlqMsg := kubemq.NewQueueMessage().
		SetChannel(channel).
		SetBody([]byte("max 3 receives, then DLQ")).
		SetMaxReceiveCount(3).
		SetMaxReceiveQueue(dlqChannel)

	if err := upstream.Send("req-dlq", []*kubemq.QueueMessage{dlqMsg}); err != nil {
		log.Fatal(err)
	}

	select {
	case res := <-upstream.Results:
		if res != nil && res.IsError {
			log.Printf("Error: %s", res.Error)
		} else {
			fmt.Printf("Sent message with DLQ policy (max 3 receives, dlq=%s)\n", dlqChannel)
		}
	case <-time.After(3 * time.Second):
		fmt.Println("Sent message (no result confirmation within timeout)")
	}
}

How It Works

1. kubemq.NewQueueMessage().SetMaxReceiveCount(3).SetMaxReceiveQueue(dlqChannel) configures the dead-letter policy at message level using the upstream stream builder.
2. upstream.Send("req-dlq", []*kubemq.QueueMessage{dlqMsg}) sends the policy-bearing message via the persistent stream; the broker stores it with the inline DLQ configuration.
3. After three failed receive attempts (nack or transaction expiry) the broker automatically routes the message to go-queues-stream.dead-letter-policy.dlq without any client action.
4. The DLQ channel is a regular queue that operators can monitor and selectively reprocess; no special consumer or channel type is needed.

Related

• Pattern overview
• Go SDK Reference
• Send & Receive
• Ack All