Ack All

Overview

AckAllQueueMessages acknowledges every pending message on a channel in a single broker-side call, without receiving them first. Reach for it when you want to drain a queue rather than process it — clearing a backlog of stale work after a bad deploy, resetting a channel between test runs, or discarding messages that are no longer relevant — where pulling and acking each message individually would be slow and wasteful.

Because it settles the whole channel at once, it is far cheaper than a receive-then-ack loop: the broker confirms all in-flight messages atomically and reports how many were affected via WaitTimeSeconds, which bounds how long it waits for in-flight transactions to settle before counting.

Gotchas: this is a blunt, irreversible instrument — it acknowledges all currently-pending messages, not a selected subset, so anything unprocessed is discarded, not redelivered. A busy channel may need a larger WaitTimeSeconds value to catch messages still landing. For routine, per-message cleanup use ordinary acks, an expiration policy, or a dead-letter queue instead — save ack-all for deliberate, wholesale purges.

Prerequisites

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

Code

// Example: queues/ack-all
//
// Demonstrates acknowledging all messages in a queue at once.
// This is useful for purging or bulk-acknowledging queue messages.
//
// Channel: go-queues.ack-all
// Client ID: go-queues-ack-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-ack-all-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

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

	// Send some messages to the queue.
	for i := 1; i <= 3; 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 3 messages")

	// Acknowledge all messages in the queue.
	ackResp, err := client.AckAllQueueMessages(ctx, &kubemq.AckAllQueueMessagesRequest{
		Channel:         channel,
		WaitTimeSeconds: 5,
	})
	if err != nil {
		log.Fatal(err)
	}
	if ackResp.IsError {
		log.Printf("Ack warning: %s", ackResp.Error)
	}
	fmt.Printf("Acknowledged %d messages\n", ackResp.AffectedMessages)
}

How It Works

1. Three messages are enqueued with client.SendQueueMessage in a loop, each identified by its result.MessageID.
2. client.AckAllQueueMessages(ctx, &kubemq.AckAllQueueMessagesRequest{...}) sends a single broker-side bulk-acknowledge that atomically confirms all pending messages on the channel without polling them one by one.
3. WaitTimeSeconds controls how long the broker waits for in-flight transactions to settle before counting the acknowledgements.
4. ackResp.AffectedMessages reports how many messages were acknowledged; ackResp.IsError signals partial failure.

Related

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