Purge Queue

Overview

Purging a queue is a management-plane operation for wiping a channel's backlog without receiving and discarding messages one at a time. Reach for it when a bad producer floods a channel, when you need a clean slate between test runs, or when you're resetting a queue during a maintenance window — all without deleting and recreating the channel itself.

AckAllQueueMessages tells the broker directly to acknowledge and drop every message still pending on the channel, entirely server-side. You give it a channel and a WaitTimeSeconds drain window so the broker can settle in-flight deliveries before finalizing, and it hands back an AffectedMessages count so you can confirm exactly how much was cleared.

Gotchas: the purge is irreversible — there's no undo once messages are acknowledged away. It only reaches messages still waiting in the queue; anything already delivered to and held by an active consumer is untouched, so a purge run right after a receive can still leave stragglers. And purging empties the channel, it doesn't delete it — new messages can be sent immediately afterward.

Prerequisites

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

Code

// Example: management/purge-queue
//
// Demonstrates purging all messages from a queue using AckAllQueueMessages.
// This effectively removes all pending messages from the queue.
//
// Channel: go-management.purge-queue
// Client ID: go-management-purge-queue-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-management-purge-queue-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-management.purge-queue"

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

	// Purge the queue by acknowledging all messages.
	purgeResp, err := client.AckAllQueueMessages(ctx, &kubemq.AckAllQueueMessagesRequest{
		Channel:         channel,
		WaitTimeSeconds: 5,
	})
	if err != nil {
		log.Fatal(err)
	}
	if purgeResp.IsError {
		log.Printf("Purge warning: %s", purgeResp.Error)
	} else {
		fmt.Printf("Purged %d messages from queue\n", purgeResp.AffectedMessages)
	}
}

How It Works

1. client.AckAllQueueMessages is a management-plane call that instructs the broker to acknowledge and discard all pending messages on the target channel without receiving them one by one.
2. WaitTimeSeconds: 5 sets the server-side drain window — the broker waits up to 5 seconds for in-flight deliveries to settle before reporting the final AffectedMessages count.
3. The return value's AffectedMessages field reports how many messages were removed, useful for auditing or confirming the purge completed.
4. This is equivalent to consuming and discarding all messages, but far more efficient for large queues since it happens entirely on the server side.

Related

• Go SDK Reference
• Create Channel
• Delete Channel