Feature Request: Per-Subscriber Prefetch Count (RabbitMQ-style QoS)

[jhoukem]

Proposed change

Summary

I would like to request a per-subscriber prefetch count feature in JetStream, similar to RabbitMQ’s basic.qos(prefetch_count=N).

Currently, JetStream allows controlling MaxAckPending, which applies per consumer. When multiple subscribers share a single durable consumer / queue group, MaxAckPending is global, and there is no way to limit how many messages a single subscriber can receive at a time.

---

Motivation / Use Case

Consider a high-throughput processing system with multiple worker instances consuming from the same queue group. Each worker can only process a limited number of messages concurrently.

• With RabbitMQ, prefetch_count ensures each subscriber receives at most N un-acked messages, giving predictable per-worker concurrency.

• In JetStream today:

  • MaxAckPending applies per consumer, not per subscriber.
  • If some subscribers go down, remaining subscribers may suddenly receive a large number of messages, overwhelming them.
  • This makes per-subscriber concurrency control difficult.

---

Proposal

• Introduce a PrefetchCount or similar option at the subscriber level.
• The broker would ensure no subscriber receives more than N un-acked messages, independent of other subscribers sharing the consumer.
• This should integrate with queue groups so each subscriber respects its own prefetch limit.

---

Attempted Workarounds

I tried to achieve this using the new JetStream Go API without success. For example, I tried limiting the number of messages fetched per pull using:

jetstream.PullMaxMessages(10)

• This only limits the client-side batch per pull request, not per-subscriber concurrency.
• Even with this, a subscriber can receive many more messages if MaxAckPending for the consumer allows it.

This illustrates that the current API does not provide the requested behavior natively.

---

Benefits

• Predictable per-subscriber concurrency
• Reduces risk of slow-consumer warnings
• Simplifies design of high-throughput workloads with variable processing times

[MauriceVanVeen]

there is no way to limit how many messages a single subscriber can receive at a time

Seems what you'd want here is a pull consumer where you're specifically in control of how many messages you pull/fetch at any given time.

Since you mentioned using jetstream.PullMaxMessages(10) I'm assuming you're using c.Consume(handler, ...). That indeed does not give you the ability to pull messages yourself, since that pulls messages automatically.

Instead what you'd want is to fetch the messages yourself using c.Fetch(10). That allows you to limit how many un-acked messages you receive, and once you've processed and acknowledged these messages, you fetch the next N un-acked messages. That seems to exactly satisfy this as well:

ensures each subscriber receives at most N un-acked messages, giving predictable per-worker concurrency.

[jhoukem]

The issue with pull consumers is that I have to finish the whole batch before fetching a new one. I could wrap it in a custom controller that fetches a new message whenever a slot becomes available, but that feels like something the library should expose directly.

There’s also the problem with NakWithDelay: delayed nacks still count as pending acks, which can eventually stall the entire subscriber once all slots are occupied by delayed messages.

What I’m trying to achieve is a pub/sub-style workflow where messages can retry with exponential backoff without blocking the rest of the flow, while guaranteeing that each subscriber always processes at most X messages in parallel and never receives more than that.