Add jetstream consumer priority groups (ADR-42) - #1015
Conversation
- Use "none" as the wire value for PriorityPolicy.NONE; the server rejects an empty string. - Send priority=0 in pull requests; it is the highest priority, and a truthiness check dropped it. - Validate min_pending/min_ack_pending/priority in fetch() so the batch path is covered too, not only _fetch_one. - Make PriorityGroupState.pinned_client_id optional and parse pinned_ts; the server omits both until a client is pinned, which made consumer_info raise on any unpinned group. - Convert priority_groups in ConsumerInfo.from_response. - Only send priority_timeout when set. - Gate overflow/pinned/unpin tests on 2.11, not 2.12.
|
Claude finished @caspervonb's task in 5m 21s —— View job PR Review: Add JetStream Consumer Priority Groups (ADR-42)
SummaryThe PR correctly addresses all the bugs called out in the description (the empty Correctness
Style
No issues found in |
![synadia-claude-reviewer[bot]](https://avatars.githubusercontent.com/u/132956691?s=60&v=4){kind=small}
| pin_id = self.pin_id | ||
| if pin_id: | ||
| next_req["id"] = pin_id | ||
| if min_pending: |
There was a problem hiding this comment.
The priority fix correctly uses if priority is not None: to handle priority=0, but min_pending and min_ack_pending still use truthiness checks here (and in two more spots in _fetch_n). Currently safe because validation rejects 0, but the inconsistency could become a silent bug if those constraints ever relax. Suggest if min_pending is not None: / if min_ack_pending is not None: throughout for consistency.
| # Any other type of status message is an error. | ||
| raise nats.js.errors.APIError.from_msg(msg) | ||
| else: | ||
| pin_id = msg.headers.get("Nats-Pin-Id") if msg.headers else None |
There was a problem hiding this comment.
"Nats-Pin-Id" is duplicated across 5 lines in _fetch_one and _fetch_n. The existing Header enum in api.py is the canonical home for NATS header names — adding PIN_ID = "Nats-Pin-Id" there would eliminate the repeated literals.
| # a possible i/o timeout error or due to a disconnection. | ||
| got_any_response = True | ||
| pass | ||
| elif JetStreamContext._is_pin_id_mismatch_error(status): |
There was a problem hiding this comment.
_fetch_one raises nats.errors.TimeoutError on PIN_ID_MISMATCH (via _is_temporary_error), whereas here in _fetch_n a PIN_ID_MISMATCH in the no-wait phase just clears the pin and falls through silently, eventually returning an empty or partial result. The behaviour is observable: fetch(1) raises on mismatch, fetch(2) does not. Raising (or at minimum documenting) the difference would avoid surprise for callers.
2 participants
Supersedes #786, which had drifted behind main. Carries @oliverlambson's original commit as-is, plus a follow-up commit addressing the review findings on that PR and a few more found while rebasing:
PriorityPolicy.NONEsent""; the server only accepts"none".priority=0(the highest priority) was dropped from pull requests by a truthiness check.min_pending/min_ack_pending/prioritywere only validated on the single-message fetch path.ConsumerInforaised on any consumer with an unpinned priority group, sincepinned_client_idwas required but the server omits it until a client is pinned.pinned_tsis now parsed as well.Closes #784