Nack All
Negative-acknowledge all received KubeMQ queue messages at once using the Go SDK to requeue them.
Overview
Bulk nack rejects an entire polled batch of queue messages in a single call instead of settling each one individually. It's the operation you reach for when a failure affects the whole batch at once — a downstream dependency is down, a shared resource lock couldn't be acquired, or a transient error means none of the messages can be processed right now — and retrying them one-by-one would just be extra round-trips for the same outcome.
It works with manual-ack polling: receiver.Poll with AutoAck: false holds the returned messages in a broker-side transaction keyed by dm.TransactionID, and resp.NackAll() sends one negative-acknowledge that settles every message in that transaction, returning them all to the queue for redelivery.
Gotchas: the receive count increments on every message in the batch, so an unbounded retry loop is one bad NackAll() away — pair it with SetMaxReceiveCount and a dead-letter policy. NackAll() is all-or-nothing: you can't use it to keep a few messages and reject the rest — that needs per-message ack/nack or a range operation. And calling it on an empty poll result is a wasted round-trip, so guard on len(resp.Messages) > 0 first.
Prerequisites
- KubeMQ server running on
localhost:50000 - Go SDK installed (
go get github.com/kubemq-io/kubemq-go/v2)
Code
// Example: queues-stream/nack-all
//
// Demonstrates rejecting all messages in a transaction using NackAll.
// Rejected messages are returned to the queue for reprocessing (up to
// the MaxReceiveCount limit, after which they are discarded).
//
// Channel: go-queues-stream.nack-all
// Client ID: go-queues-stream-nack-all-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-nack-all-client"),
)
if err != nil {
log.Fatal(err)
}
defer client.Close()
channel := "go-queues-stream.nack-all"
// Send a message with a receive policy so the server knows how to
// handle it after rejection (re-deliver up to 3 times).
_, err = client.SendQueueMessage(ctx, kubemq.NewQueueMessage().
SetChannel(channel).
SetBody([]byte("will be nacked")).
SetMaxReceiveCount(3))
if err != nil {
log.Fatal(err)
}
fmt.Println("Message sent")
// Receive the message via downstream receiver.
receiver, err := client.NewQueueDownstreamReceiver(ctx)
if err != nil {
log.Fatal(err)
}
defer receiver.Close()
resp, err := receiver.Poll(ctx, &kubemq.PollRequest{
Channel: channel,
MaxItems: 10,
WaitTimeoutSeconds: 5,
AutoAck: false,
})
if err != nil {
log.Fatal(err)
}
for _, dm := range resp.Messages {
if dm.Message != nil {
fmt.Printf("Received: body=%s tx=%s\n", dm.Message.Body, dm.TransactionID)
}
}
// NAckAll: reject all messages in the transaction (return to queue).
if len(resp.Messages) > 0 {
fmt.Printf("NAckAll: rejecting all messages from tx=%s\n", resp.Messages[0].TransactionID)
if err := resp.NackAll(); err != nil {
log.Printf("NackAll: %v", err)
}
fmt.Println("All messages rejected (returned to queue)")
}
}
How It Works
kubemq.NewQueueMessage().SetMaxReceiveCount(3)configures the message so the broker re-delivers it up to 3 times before discarding it (or routing to a DLQ if configured).receiver.Poll(ctx, &kubemq.PollRequest{AutoAck: false})holds the messages in a broker-side transaction referenced bydm.TransactionID.resp.NackAll()sends a single negative-acknowledge for the entire transaction, atomically returning all held messages to the queue for redelivery.- The receive count on each returned message is incremented by the broker; once it reaches
MaxReceiveCount, the message is discarded rather than redelivered again.
Related
Was this page helpful?