Nack All
Negative-acknowledge all received KubeMQ queue messages at once using the Python SDK poll API.
Overview
Bulk nack rejects an entire polled batch of queue messages in a single call instead of settling each one individually. It's the operation you reach for when a failure affects the whole batch at once — a downstream dependency is down, a shared resource lock couldn't be acquired, or a transient error means none of the messages can be processed right now — and retrying them one-by-one would just be extra round-trips for the same outcome.
It works with manual-ack polling: receive_queue_messages returns a response object holding the messages without settling them, and response.reject_all() sends one bulk-nack that settles every message in that response, returning them all to the queue for redelivery.
Gotchas: the receive count increments on every message in the batch, so an unbounded retry loop is one bad reject_all() away — pair it with a max-receive-count and a dead-letter policy. reject_all() is all-or-nothing: you can't use it to keep a few messages and reject the rest — that needs per-message ack/reject or a range operation. And calling it on an empty poll result is a wasted round-trip, so guard on response.is_empty() first.
Prerequisites
- KubeMQ server running on
localhost:50000 - Python SDK installed (
pip install kubemq)
Code
"""Example: Nack all — reject all received messages at once."""
from __future__ import annotations
import asyncio
from kubemq.queues import AsyncClient as AsyncQueuesClient
from kubemq import QueueMessage
async def main() -> None:
async with AsyncQueuesClient(
address="localhost:50000",
client_id="python-queues-stream-nack-all-client",
) as client:
await client.send_queue_message(
QueueMessage(
channel="python-queues-stream.nack-all",
body=b"will-be-rejected",
)
)
response = await client.receive_queue_messages(
channel="python-queues-stream.nack-all",
max_messages=10,
wait_timeout_seconds=5,
)
print(f"Received {response.count()} messages")
if not response.is_empty():
await response.reject_all()
print("All messages rejected (nacked)")
if __name__ == "__main__":
asyncio.run(main())
How It Works
response.reject_all()sends a single bulk-nack command to the broker, returning all polled messages to the queue for redelivery.response.is_empty()guards against callingreject_all()on an empty poll, which is a no-op but avoids unnecessary network round-trips.response.count()returns the number of messages in the poll response, used here for progress reporting.- Bulk nack is useful when a consumer detects a transient failure and wants to release all in-flight messages without processing any of them.
Related
Was this page helpful?