Ack Range

Overview

A single poll response often bundles several messages into one batch, but "successfully processed" rarely applies to all of them uniformly — one handler might fail while its siblings succeed. Settling the whole batch together forces an all-or-nothing outcome: either you redeliver work you already finished, or you silently drop work you didn't. Range acknowledgment lets you settle exactly the subset that actually succeeded, in one call.

KubeMQ.PollResponse.ack_range/2 takes the poll response and a list of broker-assigned sequence numbers — read from each message's attributes.sequence — and acknowledges all of them together. Any sequence you leave out of the list is left unsettled and returns to the queue for redelivery once the visibility window expires.

Gotchas: the sequences you pass must be numbers your handler actually received in that poll — passing a stale or unknown sequence isn't a safe no-op you can rely on. Messages you never include in any ack_range call aren't implicitly skipped forever; they come back for redelivery once the timeout elapses, so forgetting to settle a message isn't "done," it's "will retry." And selective settlement only works when the poll isn't using auto-ack — with auto-ack on, the broker settles the whole batch the moment it's delivered.

Prerequisites

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

Code

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

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

IO.puts("Sent 5 messages")

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

    sequences =
      poll.messages
      |> Enum.filter(& &1.attributes)
      |> Enum.map(& &1.attributes.sequence)

    ack_seqs = Enum.take(sequences, 3)
    IO.puts("Acking sequences: #{inspect(ack_seqs)}")

    case KubeMQ.PollResponse.ack_range(poll, ack_seqs) do
      :ok -> IO.puts("Range ack successful!")
      {:error, err} -> IO.puts("Ack range failed: #{err.message}")
    end

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

KubeMQ.Client.close(client)

How It Works

• PollResponse.ack_range/2 acknowledges only the specified sequence numbers
• Unacknowledged messages remain in the queue for later processing
• Sequence numbers are available via message.attributes.sequence

Related

• Ack/Reject
• Nack All