Ack/Reject

Overview

Ack and reject give you per-message control over queue delivery instead of an all-or-nothing batch outcome. When poll_queue/2 fetches a batch, the returned %PollResponse{} holds those messages in an open transaction on the broker — invisible to other consumers — until the consumer explicitly settles it. That's what you need when one bad record in a batch shouldn't take the rest down with it.

Settlement happens through calls scoped to the transaction, identified by transaction_id: acknowledging removes the settled messages from the queue permanently, while rejecting returns them to the queue for redelivery. Internally the broker tracks this against a receive count, which a dead-letter policy can use to stop retrying a poison message forever.

Gotchas: an unsettled message isn't gone — it snaps back to the queue once the transaction expires, so a slow consumer looks identical to a rejecting one; settle every message before that deadline, and never assume a batch is fully processed until you've explicitly acknowledged or rejected it via the transaction.

Prerequisites

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

Code

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

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

IO.puts("Sent 3 messages")

case KubeMQ.Client.poll_queue(client,
       channel: channel,
       max_items: 3,
       wait_timeout: 5_000
     ) do
  {:ok, poll} ->
    IO.puts("Polled #{length(poll.messages)} messages")
    IO.puts("Transaction ID: #{poll.transaction_id}")

    Enum.each(poll.messages, fn msg ->
      IO.puts("  #{msg.body}")
    end)

    {good, bad} = Enum.split(Enum.map(poll.messages, & &1.attributes.sequence), 2)

    case KubeMQ.PollResponse.ack_range(poll, good) do
      :ok -> IO.puts("Acknowledged sequences: #{inspect(good)}")
      {:error, err} -> IO.puts("Ack failed: #{err.message}")
    end

    case KubeMQ.PollResponse.nack_range(poll, bad) do
      :ok -> IO.puts("Rejected sequences: #{inspect(bad)}")
      {:error, err} -> IO.puts("Reject failed: #{err.message}")
    end

  {:error, err} ->
    IO.puts("Poll failed: #{err.message}")
end

KubeMQ.Client.close(client)

How It Works

• poll_queue/2 returns a %PollResponse{} with transactional control
• PollResponse.ack_range/2 acknowledges only the sequences you pass it, leaving the rest of the batch pending
• PollResponse.nack_range/2 rejects a chosen subset for redelivery, independent of the sequences you acked
• The transaction_id uniquely identifies the poll transaction

Related

• Ack All
• Nack All
• Ack Range