[improve][pip] PIP-493: Key-ordered negative acknowledgement for Key_Shared subscriptions - #26418
Open
lhotari wants to merge 2 commits into
Open
[improve][pip] PIP-493: Key-ordered negative acknowledgement for Key_Shared subscriptions#26418lhotari wants to merge 2 commits into
lhotari wants to merge 2 commits into
Conversation
…Shared subscriptions
PIP-379 defined the Key_Shared ordering contract as an invariant, and carved
negative acknowledgement out of it: messages scheduled for redelivery by
negativeAcknowledge "will get redelivered as soon as possible" with "no ordering
guarantee in these cases". PIP-493 replaces that carve-out with a guarantee.
The problem is that negativeAcknowledge, the natural tool for "try this again in a
moment" after a transient downstream failure, silently reorders a key. Later
messages for the key are already in the consumer's receiver queue and get processed
while the nacked one waits out its retry delay. The alternatives are worse: a dead
letter topic leaves a permanent gap in the key's sequence, and blocking the consumer
thread stalls every other key it handles. An application cannot fix this itself,
because only the broker can stop dispatching messages for a key.
The design has two halves, because at the moment of the nack the messages that must
not reach the application are in two places.
On the broker, a new CommandNegativeAck is sent at nack time and marks the message's
sticky key hash *retrying*. While a hash is retrying, only the earliest negatively
acknowledged position of that hash may be dispatched -- that position is the retry
itself -- and the rule is applied at all three existing dispatch gates. The command
carries no delay: the retry timer, RedeliveryBackoff and the acknowledgement grouping
window all stay on the client exactly as they are today, so the broker gains no
scheduler and the nacked message stays inside the existing unacknowledged-message
accounting.
On the client, messages for the guarded key that are already inside the consumer are
*returned* to the broker rather than held. Holding them would leave them in the
consumer's PendingAcksMap, so a hash reassignment would make PIP-379 mark the hash
draining until they are acknowledged, while the client refuses to hand them over
until the nacked message is acknowledged -- a circular wait broken only by an
acknowledgement timeout, which is disabled by default. Returning dissolves the cycle
and removes the need for any client-side buffer. Returned messages travel in a
separate list so their redelivery count is not incremented, because the application
never saw them.
The blocking rule is position-ordered rather than key-set membership, at all three
places it appears -- the broker gate, the client guard and the admission API. A
message is withheld exactly when an earlier position of the same key is still
unresolved. Set membership looks equivalent and is not: it also withholds positions
earlier than the nacked one, which wedges permanently on the broker and prematurely
disarms the guard on the client.
Batch-index acknowledgement is handled end to end, since it is enabled by default and
a dispatched entry carries the broker's batch-index state. A negatively acknowledged
batched entry populates the existing MessageIdData batch_index and ack_set -- no
protobuf schema change -- and the client flushes its pending grouped acknowledgements
before sending the command, so the broker observes which indexes are acknowledged
rather than assuming it.
The client API additions are opt-in. OrderedNegativeAckMode {DISABLED, BEST_EFFORT,
REQUIRED} lets an application that depends on the guarantee fail fast against a
broker that does not advertise support, while the default BEST_EFFORT preserves
today's silent fallback. Consumer#acquireForOrderedProcessing returns an
OrderedProcessingAdmission for applications that hold several messages in flight
behind per-key queues, so the library rather than the application adjudicates whether
a queued message may be processed. A client-side broad-failure delivery pause answers
the common case where the downstream is failing for every key regardless of key, by
stopping flow permits instead of opening as many concurrent retry cycles as there are
active keys.
Nothing is persisted, no metadata or serialized format changes, and the protocol
change is additive. Old clients are unaffected; new clients against an old broker
fall back to today's behaviour.
Assisted-by: Claude Code (Opus 5)
2 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Motivation
I'd like to propose PIP-493, which makes negative acknowledgement
usable on Key_Shared subscriptions without breaking key ordering.
PIP-379 defined the Key_Shared ordering contract as an invariant, and
then carved negative acknowledgement out of it: messages scheduled for
redelivery by negativeAcknowledge "will get redelivered as soon as
possible", with "no ordering guarantee in these cases". PIP-493
replaces that carve-out with a guarantee.
The gap matters because negativeAcknowledge is the natural tool for
"try this again in a moment" after a transient downstream failure --
an HTTP 503, a database failover, a rate limit -- and on a Key_Shared
subscription it silently reorders the key. Later messages for that key
are already in the consumer's receiver queue and get processed while
the nacked one waits out its retry delay. The alternatives are worse:
a dead letter topic leaves a permanent gap in the key's sequence, and
blocking the consumer thread stalls every other key that consumer
handles. An application cannot fix this on its own, because only the
broker can stop dispatching messages for a key.
The design has two halves, because at the moment of the nack the
messages that must not reach the application are in two places.
On the broker, a new CommandNegativeAck is sent at nack time and marks
the message's sticky key hash "retrying". While a hash is retrying,
only the earliest negatively acknowledged position of that hash may be
dispatched -- that position is the retry itself -- and the rule is
applied at all three existing dispatch gates. The command carries no
delay: the retry timer, RedeliveryBackoff and the acknowledgement
grouping window all stay on the client exactly as they are today, so
the broker gains no scheduler and the nacked message stays inside the
existing unacknowledged-message accounting.
On the client, messages for the guarded key that are already inside
the consumer are returned to the broker rather than held, and their
redelivery count is not incremented because the application never saw
them. Holding them instead would leave them in that consumer's pending
acknowledgements, so a hash reassignment would make PIP-379 mark the
hash draining until they are acknowledged, while the client refuses to
hand them over until the nacked message is acknowledged -- a circular
wait broken only by an acknowledgement timeout, which is disabled by
default.
The client API additions are opt-in. OrderedNegativeAckMode lets an
application that depends on the guarantee fail fast against a broker
that does not advertise support, while the default preserves today's
silent fallback. Consumer#acquireForOrderedProcessing serves
applications that keep several messages in flight behind per-key
queues, so the library rather than the application decides whether a
queued message may still be processed. A client-side broad-failure
delivery pause answers the common case where the downstream is failing
for every key rather than for one, by stopping flow permits instead of
opening as many concurrent retry cycles as there are active keys.
Nothing is persisted, no metadata or serialized format changes, and
the protocol change is additive. Old clients are unaffected; new
clients against an old broker fall back to today's behaviour.
The full proposal can be found at: #26418
Rendered PIP document:
https://github.com/lhotari/pulsar/blob/lh-pip-493-keyshared-nack-ordering/pip/pip-493.md
I welcome your feedback and discussion on this proposal. Please share
your thoughts, concerns, or suggestions.
Discussion thread: https://lists.apache.org/thread/wz6xpyoqghz5z3cdchgm9skdplytvmks