Add jetstream consumer priority groups (ADR-42) #786
Conversation
oliverlambson
force-pushed
the
js-consumer-groups
branch
from
60c24c0 to
424c353
Compare
oliverlambson
force-pushed
the
js-consumer-groups
branch
from
424c353 to
1181018
Compare
|
#788 will fix this |
|
@wallyqs @caspervonb any thoughts on this? |
oliverlambson
force-pushed
the
js-consumer-groups
branch
3 times, most recently
from
e07767b to
33960e4
Compare
This PR added omitempty to a bunch of StreamConfig fields: nats-io/nats-server#7390
oliverlambson
force-pushed
the
js-consumer-groups
branch
from
33960e4 to
1c955ba
Compare
oliverlambson
changed the title
|
Sketched out an implementation as-well, will review shortly. |
|
@caspervonb great👍 not wedded to any particular implementation, I’m just keen for the feature so I can do better event sourcing |
There was a problem hiding this comment.
Pull request overview
This PR implements support for JetStream consumer priority groups (ADR-42), enabling flexible failover and priority management for multiple clients pulling from the same consumer. The implementation is ported from nats.go and adapted to fit nats.py conventions.
Key Changes
- Adds three priority policies (PINNED, OVERFLOW, PRIORITIZED) for managing consumer message delivery
- Implements pin ID tracking to support pinned consumer behavior
- Extends fetch API with min_pending, min_ack_pending, and priority parameters for overflow and prioritized policies
- Adds unpin_consumer API for manual unpinning of pinned consumers
Reviewed changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 11 comments.
Show a summary per file
| File | Description |
|---|---|
| nats/src/nats/js/api.py | Adds PriorityPolicy enum, PriorityGroupState dataclass, PIN_ID_MISMATCH status code, and extends ConsumerConfig with priority fields |
| nats/src/nats/js/errors.py | Adds PinIdMismatchError exception for handling 423 errors |
| nats/src/nats/js/client.py | Implements priority_group parameter, pin_id tracking, and extends fetch() with priority-related parameters |
| nats/src/nats/js/manager.py | Adds unpin_consumer method for manually unpinning consumers |
| nats/tests/test_js.py | Updates version-specific assertions and adds comprehensive tests for all three priority policies |
Comments suppressed due to low confidence (3)
nats/src/nats/js/client.py:664
- Missing documentation for new parameter: The docstring should document the new
priority_groupparameter. This is an important parameter for the priority groups feature.
Suggested addition:
:param priority_group: Priority group name for this pull subscription (requires consumer configured with priority_groups).
"""
pull_subscribe returns a `PullSubscription` that can be delivered messages
from a JetStream pull based consumer by calling `sub.fetch`.
::
import asyncio
import nats
async def main():
nc = await nats.connect()
js = nc.jetstream()
await js.add_stream(name='mystream', subjects=['foo'])
await js.publish('foo', b'Hello World!')
msgs = await sub.fetch()
msg = msgs[0]
await msg.ack()
await nc.close()
if __name__ == '__main__':
asyncio.run(main())
"""
nats/src/nats/js/client.py:580
- Missing documentation for new parameter: The docstring should document the new
priority_groupparameter. This is an important parameter for the priority groups feature.
Suggested addition:
:param priority_group: Priority group name for this pull subscription (requires consumer configured with priority_groups).
"""Create consumer and pull subscription.
1. Find stream name by subject if `stream` is not passed.
2. Create consumer with the given `config` if not created.
3. Call `pull_subscribe_bind`.
::
import asyncio
import nats
async def main():
nc = await nats.connect()
js = nc.jetstream()
await js.add_stream(name='mystream', subjects=['foo'])
await js.publish('foo', b'Hello World!')
sub = await js.pull_subscribe('foo', stream='mystream')
msgs = await sub.fetch()
msg = msgs[0]
await msg.ack()
await nc.close()
if __name__ == '__main__':
asyncio.run(main())
"""
nats/src/nats/js/client.py:1086
- Missing documentation for new parameters: The docstring should document the new
min_pending,min_ack_pending, andpriorityparameters that were added to thefetchmethod. These are important parameters for priority group functionality and should be documented for API consumers.
Suggested additions:
:param min_pending: Minimum number of messages pending (OVERFLOW policy).:param min_ack_pending: Minimum number of pending acks (OVERFLOW policy).:param priority: Priority level 0-9 where 0 is highest (PRIORITIZED policy).
"""
fetch makes a request to JetStream to be delivered a set of messages.
:param batch: Number of messages to fetch from server.
:param timeout: Max duration of the fetch request before it expires.
:param heartbeat: Idle Heartbeat interval in seconds for the fetch request.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| priority_policy: Optional[PriorityPolicy] = None | ||
|
|
||
| # The duration (seconds) after which the client will be unpinned if no new | ||
| # pull requests are sent.Used with PriorityPolicy.PINNED. |
There was a problem hiding this comment.
Missing space in comment. Should be "sent. Used" instead of "sent.Used".
| # pull requests are sent.Used with PriorityPolicy.PINNED. | |
| # pull requests are sent. Used with PriorityPolicy.PINNED. |
| next_req["min_pending"] = min_pending | ||
| if min_ack_pending: | ||
| next_req["min_ack_pending"] = min_ack_pending | ||
| if priority: |
There was a problem hiding this comment.
Bug: The condition if priority: will not add priority to the request when priority=0, which is the highest priority value. This should be if priority is not None: to match the validation logic and ensure priority=0 is correctly sent to the server.
| if priority: | |
| if priority is not None: |
| next_req["min_pending"] = min_pending | ||
| if min_ack_pending: | ||
| next_req["min_ack_pending"] = min_ack_pending | ||
| if priority: |
There was a problem hiding this comment.
Bug: The condition if priority: will not add priority to the request when priority=0, which is the highest priority value. This should be if priority is not None: to match the validation logic and ensure priority=0 is correctly sent to the server.
| if priority: | |
| if priority is not None: |
| pulling from the same consumer | ||
| Introduced in nats-server 2.12.0. |
There was a problem hiding this comment.
Documentation inconsistency: The class docstring states "Introduced in nats-server 2.12.0" but PINNED and OVERFLOW were actually introduced in nats-server 2.11.0 (as mentioned in the references and test skip messages). Only PRIORITIZED was introduced in 2.12.0. Consider updating the class-level docstring to indicate "Introduced in nats-server 2.11.0" and keep the specific note about PRIORITIZED being from 2.12.0.
| pulling from the same consumer | |
| Introduced in nats-server 2.12.0. | |
| pulling from the same consumer. | |
| Introduced in nats-server 2.11.0. | |
| The PRIORITIZED policy was added in nats-server 2.12.0. |
| nc = await nats.connect() | ||
|
|
||
| server_version = nc.connected_server_version | ||
| if server_version.major == 2 and server_version.minor < 12: |
There was a problem hiding this comment.
Version check inconsistency: The skip message says "requires nats-server v2.11.0 or later" but the condition checks for < 12. This should be server_version.minor < 11 to match the documented requirement. The OVERFLOW priority policy was introduced in nats-server 2.11.0, not 2.12.0.
| if server_version.major == 2 and server_version.minor < 12: | |
| if server_version.major == 2 and server_version.minor < 11: |
|
|
||
| async def unpin_consumer(self, stream_name: str, consumer_name: str, group: str) -> None: | ||
| """ | ||
| unpin_consumer unpins a pinned consumer. |
There was a problem hiding this comment.
Incomplete documentation: The docstring should document the parameters and provide more detail about what unpinning means and when it's used.
Suggested improvement:
"""
Unpin a pinned consumer for a priority group.
Manually unpins a consumer that was pinned by the PINNED priority policy,
allowing other consumers in the same priority group to receive messages
without waiting for the pin timeout to expire.
:param stream_name: Name of the stream.
:param consumer_name: Name of the consumer.
:param group: Priority group name to unpin.
:raises NotFoundError: If the stream or consumer doesn't exist.
"""| unpin_consumer unpins a pinned consumer. | |
| Unpin a pinned consumer for a priority group. | |
| Manually unpins a consumer that was pinned by the PINNED priority policy, | |
| allowing other consumers in the same priority group to receive messages | |
| without waiting for the pin timeout to expire. | |
| :param stream_name: Name of the stream. | |
| :param consumer_name: Name of the consumer. | |
| :param group: Priority group name to unpin. | |
| :raises NotFoundError: If the stream or consumer doesn't exist. |
| next_req["min_pending"] = min_pending | ||
| if min_ack_pending: | ||
| next_req["min_ack_pending"] = min_ack_pending | ||
| if priority: |
There was a problem hiding this comment.
Bug: The condition if priority: will not add priority to the request when priority=0, which is the highest priority value. This should be if priority is not None: to match the validation logic at line 1138 and ensure priority=0 is correctly sent to the server.
| if priority: | |
| if priority is not None: |
| nc = await nats.connect() | ||
|
|
||
| server_version = nc.connected_server_version | ||
| if server_version.major == 2 and server_version.minor < 12: |
There was a problem hiding this comment.
Version check inconsistency: The skip message says "requires nats-server v2.11.0 or later" but the condition checks for < 12. This should be server_version.minor < 11 to match the documented requirement. The PINNED priority policy was introduced in nats-server 2.11.0, not 2.12.0.
| if server_version.major == 2 and server_version.minor < 12: | |
| if server_version.major == 2 and server_version.minor < 11: |
| for i in range(0, 100): | ||
| await js.publish("foo", f"{i}".encode()) | ||
| with pytest.raises(TimeoutError): | ||
| msgs = await psub.fetch(10, timeout=0.5, min_pending=110) |
| priority_group="A", | ||
| ) | ||
| with pytest.raises(TimeoutError): | ||
| msgs = await psub.fetch(10, timeout=0.5, min_ack_pending=10) |
2 participants
Stacked on #788
Closes #784
Implements support for ADR-42
This is mostly a port of the implementation in nats.go – but adapted to try be in-line with the way things are done in nats.py.