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.

purgeQueuesChannel tells the broker directly to acknowledge and drop every message still pending on the channel, entirely server-side. It takes just the channel name and returns the count of messages purged, so you can confirm exactly how much backlog was cleared.

Gotchas: the purge is irreversible — there's no undo once messages are dropped. 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
• Kotlin SDK installed (implementation("io.kubemq.sdk:kubemq-sdk-kotlin:1.0.1"))

Code

package io.kubemq.sdk.examples.management

import io.kubemq.sdk.client.KubeMQClient
import io.kubemq.sdk.queues.queueMessage
import kotlinx.coroutines.runBlocking

private const val ADDRESS = "localhost:50000"
private const val CLIENT_ID = "kotlin-management-purge-queue"
private const val CHANNEL = "kotlin-management.purge-queue"

fun main() = runBlocking {
    val client = KubeMQClient.queues {
        address = ADDRESS
        clientId = CLIENT_ID
    }

    client.use {
        try {
            client.createQueuesChannel(CHANNEL)

            // Send several messages
            for (i in 0 until 5) {
                client.sendQueuesMessage(queueMessage {
                    channel = CHANNEL
                    body = "purge-msg-$i".toByteArray()
                })
            }
            println("Sent 5 messages to queue")

            // Purge all messages from the queue
            val purged = client.purgeQueuesChannel(CHANNEL)
            println("Purged queue: $CHANNEL ($purged messages)")

            // Verify empty
            val response = client.receiveQueuesMessages {
                channel = CHANNEL
                maxItems = 10
                waitTimeoutMs = 2000
                autoAck = true
            }
            println("After purge: ${response.messages.size} messages")
        } finally {
            try { client.deleteQueuesChannel(CHANNEL) } catch (_: Exception) {}
        }
        println("Done.")
    }
}

How It Works

• createQueuesChannel creates the queue channel before use.
• sendQueuesMessage sends test messages to the queue.
• purgeQueuesChannel removes all pending messages from the queue and returns the count of purged messages.
• A follow-up receiveQueuesMessages with a short waitTimeoutMs confirms the queue is empty after the purge.
• The channel is deleted in the finally block to clean up.

Related

• Create Channel
• Delete Channel
• List Channels