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. Per-message settlement lets each message's outcome reflect what actually happened to it, instead of the worst result in the batch.

Each message returned by receive_queue_messages carries its own broker-assigned msg.sequence. Calling msg.async_ack() on one message settles only that message; calling msg.async_nack() on another explicitly returns it to the queue for redelivery. Messages you don't touch at all are left pending — still redeliverable — until their own async_ack()/async_nack() is called or the underlying transaction's visibility window expires.

Gotchas: messages you never touch aren't automatically fine — once the visibility timeout elapses, anything left unsettled goes back to the queue, so a handler that forgets to settle a message isn't "done," it's "will retry." Every nack increments that message's receive_count, which can trigger dead-letter routing if max_receive_count is configured — so a bug that nacks indiscriminately can drain a message's retry budget fast. And selective settlement only works when auto-ack is disabled; with it on, the broker settles the entire batch the moment it's delivered, before your handler even runs.

Prerequisites

• KubeMQ server running on localhost:50000
• Python SDK installed (pip install kubemq)

Code

"""Example: Ack range — selectively acknowledge messages by sequence."""

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-queues-stream-ack-range-client",
    ) as client:
        for i in range(3):
            await client.send_queue_message(
                QueueMessage(
                    channel="python-queues-stream.ack-range",
                    body=f"msg-{i}".encode(),
                )
            )

        response = await client.receive_queue_messages(
            channel="python-queues-stream.ack-range",
            max_messages=3,
            wait_timeout_seconds=10,
        )
        print(f"Received {len(response.messages)} messages")
        for msg in response.messages:
            if msg.sequence % 2 == 0:
                await msg.async_ack()
                print(f"  Acked: seq={msg.sequence}")
            else:
                await msg.async_nack()
                print(f"  Rejected: seq={msg.sequence}")


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

How It Works

• Messages are received without auto_ack, so each one must be settled explicitly via msg.async_ack() or msg.async_nack().
• Settlement decisions are made per-message using the broker-assigned msg.sequence (even = ack, odd = nack in this example).
• Nacked messages are returned to the queue for redelivery; their receive_count increments, which can trigger DLQ routing if max_receive_count is set.
• This per-message settlement pattern is the foundation for selective processing: only confirmed-successful messages are removed from the queue.

Related

• Pattern overview
• Python SDK Reference
• Send & Receive
• Ack All