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.

ack_all_queue_messages 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 wait_time_seconds drain window so the broker can settle in-flight deliveries before finalizing, and it hands back the count of acknowledged messages 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
• Python SDK installed (pip install kubemq)

Code

"""Example: Purge queue — remove all messages from a queue."""

from __future__ import annotations

import asyncio

from kubemq import AsyncQueuesClient
from kubemq import QueueMessage


async def main() -> None:
    async with AsyncQueuesClient(
        address="localhost:50000",
        client_id="python-management-purge-queue-client",
    ) as client:
        # Send some messages to purge
        for i in range(5):
            await client.send_queue_message(
                QueueMessage(
                    channel="python-management.purge-queue",
                    body=f"Msg-{i + 1}".encode(),
                )
            )
        print("Sent 5 messages")

        # Purge all messages from the queue
        acked = await client.ack_all_queue_messages(
            "python-management.purge-queue", wait_time_seconds=5
        )
        print(f"Purged {acked} messages from 'python-management.purge-queue'")


if __name__ == "__main__":
    asyncio.run(main())

How It Works

ack_all_queue_messages is used here as a purge: it pulls and acknowledges every pending message in the queue within the wait_time_seconds window, effectively draining it. The method returns the count of messages acknowledged. This is the recommended Python approach since there is no dedicated purge API in the Python SDK; the result is equivalent — the queue is empty after the call.

Related

• Python SDK Reference
• Create Channel
• Delete Channel