Ack Range
Acknowledge a range of KubeMQ queue messages by sequence number using the Go SDK stream API.
Overview
A single poll response often bundles several messages into one transaction, but "successfully processed" rarely applies to all of them uniformly — one handler might fail while its siblings succeed. Settling the whole batch together forces an all-or-nothing outcome: either you redeliver work you already finished, or you silently drop work you didn't. Per-message settlement lets each message's outcome reflect what actually happened to it, instead of the worst result in the batch.
Each DownstreamMessage returned by receiver.Poll carries its own Sequence and shares a TransactionID with the rest of the batch. Calling dm.Ack() on one message settles only that message; every other message in the same transaction is left untouched — still pending, still redeliverable — until its own Ack() or Nack() is called, or the transaction's visibility window expires.
Gotchas: messages you never touch aren't automatically fine — once the transaction times out, anything left unsettled goes back to the queue for redelivery, so a handler that forgets to call Ack() isn't "done," it's "will retry." Selective settlement only works with AutoAck: false on the poll request; with auto-ack on, the broker settles the entire batch the moment it's delivered, before your handler even runs. And there's no bulk "ack everything except these" call — tracking which sequences you've already settled is on you.
Prerequisites
- KubeMQ server running on
localhost:50000 - Go SDK installed (
go get github.com/kubemq-io/kubemq-go/v2)
Code
// Example: queues-stream/ack-range
//
// Demonstrates selectively acknowledging specific messages using individual
// message Ack/Nack methods. This allows fine-grained control over which
// messages in a transaction are acknowledged.
//
// Channel: go-queues-stream.ack-range
// Client ID: go-queues-stream-ack-range-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-ack-range-client"),
)
if err != nil {
log.Fatal(err)
}
defer client.Close()
channel := "go-queues-stream.ack-range"
// Send multiple messages via upstream stream.
upstream, err := client.QueueUpstream(ctx)
if err != nil {
log.Fatal(err)
}
defer upstream.Close()
for i := range 3 {
msg := kubemq.NewQueueMessage().
SetChannel(channel).
SetBody(fmt.Appendf(nil, "msg-%d", i))
if err := upstream.Send(fmt.Sprintf("req-%d", i), []*kubemq.QueueMessage{msg}); err != nil {
log.Fatal(err)
}
// Drain result
select {
case <-upstream.Results:
case <-time.After(3 * time.Second):
}
}
fmt.Println("Sent 3 messages")
// Allow messages to be committed to the queue.
time.Sleep(time.Second)
// Receive messages 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 seq=%d\n", dm.Message.Body, dm.Sequence)
}
}
// Selectively ack only the first message.
if len(resp.Messages) > 0 {
dm := resp.Messages[0]
fmt.Printf("Ack: acking sequence %d from tx=%s\n", dm.Sequence, dm.TransactionID)
if err := dm.Ack(); err != nil {
log.Printf("Ack failed: %v", err)
}
fmt.Println("Selective ack complete")
}
}
How It Works
- Messages are sent via
upstream.Sendin a loop, each with a uniquerefRequestID; results are drained immediately fromupstream.Resultsto prevent the channel from filling. receiver.Poll(ctx, &kubemq.PollRequest{AutoAck: false})fetches all queued messages into a single transaction; eachDownstreamMessagecarries its ownSequenceandTransactionID.dm.Ack()on a single message settles only that message within the transaction; the remaining messages in the same transaction are left unsettled (they will return to the queue when the transaction expires).- This per-message settlement enables selective processing: acknowledge only the messages your handler successfully processed and leave the rest for redelivery.
Related
Was this page helpful?