As you may know, RabbitMQ employs a rather extensive Acknowledgement policy.
In most cases, FastStream automatically acknowledges (acks) messages on your behalf. When your function executes correctly, including sending all responses, a message will be acknowledged (and rejected in case of an exception).
However, there are situations where you might want to use a different acknowledgement logic.
If you prefer to use a nack instead of a reject when there's an error in message processing, you can specify the
retry flag in the
@broker.subscriber(...) method, which is responsible for error handling logic.
By default, this flag is set to
False, indicating that if an error occurs during message processing, the message can still be retrieved from the queue:
If this flag is set to
True, the message will be nacked and placed back in the queue each time an error occurs. In this scenario, the message can be processed by another consumer (if there are several of them) or by the same one:
retry flag is set to an
int, the message will be placed back in the queue, and the number of retries will be limited to this number:
At the moment, attempts are counted only by the current consumer. If the message goes to another consumer, it will have its own counter. Subsequently, this logic will be reworked.
For more complex error handling cases, you can use tenacity
If you want to acknowledge a message manually, you can get access directly to the message object via the Context and call the method.
FastStream will see that the message was already acknowledged and will do nothing at process end.
If you want to interrupt message processing at any call stack, you can raise
This way, FastStream interrupts the current message processing and acknowledges it immediately. Also, you can raise
If you want to disable FastStream Acknowledgement logic at all, you can use
@broker.subscriber(..., no_ack=True) option. This way you should always process a message (ack/nack/terminate/etc) by yourself.