Ack & Reject

Overview

Ack and reject give you per-message control over queue delivery instead of an all-or-nothing batch outcome. When receiver.Poll fetches a batch with AutoAck: false, each message stays in an open transaction on the broker — invisible to other consumers — until the consumer explicitly settles it. That's what you need when one bad record in a batch shouldn't take the rest down with it.

Settlement happens through two calls on the DownstreamMessage: dm.Ack(), which permanently removes the message from the queue, and dm.Nack(), which returns it to the queue for redelivery. Internally the broker tracks this against a receive count, which a dead-letter policy can use to stop retrying a poison message forever.

Gotchas: an unsettled message isn't gone — it snaps back to the queue once the transaction expires, so a slow consumer looks identical to a rejecting one; settle every message before that deadline, and never assume a batch is fully processed until you've called Ack() or Nack() on each one individually.

Prerequisites

• KubeMQ server running on localhost:50000
• Go SDK installed (go get github.com/kubemq-io/kubemq-go/v2)

Code

// Example: queues/ack-reject
//
// Demonstrates individual message acknowledgment and rejection using
// the queue downstream receiver. Messages can be individually acked
// (confirmed) or rejected (nacked) back to the queue.
//
// Channel: go-queues.ack-reject
// Client ID: go-queues-ack-reject-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-ack-reject-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-queues.ack-reject"

	// Send two messages.
	for i := 1; i <= 2; i++ {
		_, err := client.SendQueueMessage(ctx, kubemq.NewQueueMessage().
			SetChannel(channel).
			SetBody(fmt.Appendf(nil, "msg-%d", i)))
		if err != nil {
			log.Fatal(err)
		}
	}
	fmt.Println("Sent 2 messages")

	// Poll messages with manual acknowledgment (AutoAck=false).
	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)
	}

	fmt.Printf("Polled %d messages\n", len(resp.Messages))
	for _, dm := range resp.Messages {
		if dm.Message != nil {
			fmt.Printf("  body=%s\n", dm.Message.Body)
		}
	}
}

How It Works

1. client.NewQueueDownstreamReceiver(ctx) creates a long-lived downstream gRPC stream; unlike PollQueue, it is reusable across multiple Poll calls and kept open by defer receiver.Close().
2. receiver.Poll(ctx, &kubemq.PollRequest{AutoAck: false}) fetches up to MaxItems messages and holds them in an open transaction on the broker — they remain invisible to other consumers until the transaction is settled.
3. With AutoAck: false, messages must be settled by calling dm.Ack() (removes from queue), dm.Nack() (returns for redelivery), or by letting the transaction expire (also returns for redelivery).
4. The DownstreamMessage.TransactionID field identifies the transaction; individual dm.Ack() / dm.Nack() calls settle exactly one message in that transaction.

Related

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