📝 Feature: Support broker-level ack_policy default with per-subscriber override (#2826)
Problem
Currently
ack_policy can only be set per-subscriber:@broker.subscriber("orders.topic", ack_policy=AckPolicy.NACK_ON_ERROR)
async def handle_order(msg: OrderMessage) -> None:
...
There is no way to set a default
ack_policy at the broker (or router) level. In practice, most services want the same policy across all subscribers — typically NACK_ON_ERROR for at-least-once delivery with DLQ. This forces every @broker.subscriber() call to repeat the same ack_policy= argument, which is error-prone: forgetting it on a single subscriber silently falls back to ACK_FIRST (at-most-once), which can cause silent message loss.Proposed solution
Add an
ack_policy parameter to KafkaBroker (and equivalently to RabbitBroker, NatsBroker, etc.) that sets the default for all subscribers registered on that broker. Individual subscribers can still override it.broker = KafkaBroker("localhost:9092", ack_policy=AckPolicy.NACK_ON_ERROR)
# Inherits NACK_ON_ERROR from broker
@broker.subscriber("orders.topic")
async def handle_order(msg: OrderMessage) -> None:
...
# Overrides to ACK for this specific subscriber
@broker.subscriber("notifications.topic", ack_policy=AckPolicy.ACK)
async def handle_notification(msg: NotificationMessage) -> None:
...
The resolution order would be: subscriber-level > broker-level > built-in default (
ACK_FIRST).Why this matters
• Safety:
ACK_FIRST as a silent default is dangerous for services that use DLQ or need at-least-once delivery. A broker-level default lets teams enforce their delivery guarantee in one place.• DRY: Services with 10+ subscribers shouldn't need to repeat
ack_policy=AckPolicy.NACK_ON_ERROR on every one.• Consistency with other broker-level settings:
decoder, middlewares, security, and logger are all broker-level defaults that subscribers inherit. ack_policy is the notable exception.#enhancement #good_first_issue #faststream #ag2ai
sent via relator