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.

purge_queue_channel/2 tells the broker directly to acknowledge and drop every message still pending on the specified channel, entirely server-side, returning :ok on success. Following up with a poll_queue call confirms the queue is empty afterward.

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. Large queues can also make the purge call itself take a while, so a timeout on the request is worth handling explicitly rather than assuming instant completion.

Prerequisites

• KubeMQ server running on localhost:50000
• Elixir SDK installed ({:kubemq, "~> 1.0"} in mix.exs)

Code

channel = "elixir-mgmt.purge-queue"
{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "elixir-mgmt-purge")

for i <- 1..10 do
  {:ok, _} = KubeMQ.Client.send_queue_message(client,
    KubeMQ.QueueMessage.new(channel: channel, body: "Purgeable #{i}"))
end

IO.puts("Sent 10 messages to '#{channel}'")
Process.sleep(1_000)

try do
  case KubeMQ.Client.purge_queue_channel(client, channel) do
    :ok -> IO.puts("Queue purged successfully!")
    {:error, err} -> IO.puts("Purge failed: #{err.message}")
  end
catch
  :exit, {:timeout, _} ->
    IO.puts("Purge timed out (server may still be processing)")
end

case KubeMQ.Client.poll_queue(client,
       channel: channel,
       max_items: 100,
       wait_timeout: 2_000
     ) do
  {:ok, poll} ->
    IO.puts("Messages after purge: #{length(poll.messages)}")
    if poll.messages != [], do: KubeMQ.PollResponse.ack_all(poll)

  {:error, _} ->
    IO.puts("Queue is empty (confirmed)")
end

KubeMQ.Client.close(client)

How It Works

• purge_queue_channel/2 removes all pending messages from the specified queue
• A poll after purging confirms the queue is empty
• The try/catch handles potential timeout errors for large queues

Related

• Ack All
• List Channels