Ack & Reject

Overview

Ack and reject give you per-message control over queue delivery instead of an all-or-nothing batch outcome. When receive_queue_messages fetches a batch without auto_ack=True, each message stays 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 two calls on the received message: msg.async_ack(), which permanently removes it from the queue, and msg.async_nack(), which returns it to the queue for redelivery. Internally the broker tracks this against a receive count (max_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 called async_ack() or async_nack() on each one individually.

Prerequisites

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

Code

"""Example: Ack and reject — demonstrate per-message ack and reject decisions."""

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-ack-reject-client",
    ) as client:
        # Send test messages
        for i in range(4):
            await client.send_queue_message(
                QueueMessage(
                    channel="python-queues.ack-reject",
                    body=f"Order-{i + 1}".encode(),
                )
            )
        print("Sent 4 messages")

        # Receive and selectively ack or reject
        response = await client.receive_queue_messages(
            channel="python-queues.ack-reject",
            max_messages=4,
            wait_timeout_seconds=10,
        )
        for msg in response.messages:
            body = msg.body.decode("utf-8")
            # Simulate processing: ack even-numbered, reject odd-numbered
            if msg.sequence % 2 == 0:
                await msg.async_ack()
                print(f"  Acked: {body} (seq={msg.sequence})")
            else:
                await msg.async_nack()
                print(f"  Rejected: {body} (seq={msg.sequence})")


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

How It Works

• receive_queue_messages returns all messages in a single poll; without auto_ack=True, each message must be settled individually.
• msg.async_ack() tells the broker the message was successfully processed and removes it from the queue permanently.
• msg.async_nack() tells the broker processing failed; the message is returned to the queue for redelivery (or routed to DLQ if max_receive_count is exceeded).
• The msg.sequence field is a monotonically increasing broker-assigned integer used here to distinguish odd from even messages.

Related

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