[improve][pip] PIP-491: Prevent Delivery Stalls by Making the Client Return Exactly the Permits Used by the Broker

[void-ptr974]

Motivation

For every delivered command, the broker uses a number of consumer permits and the client must eventually return exactly that number. Pulsar does not currently send this count on the wire. Instead, broker layers and the client derive related counts independently from the information available at different stages.

Those calculations agree during normal complete, partial, and non-batched delivery. They can disagree when admission changes the final send set, payload processing fails before or during batch expansion, a message has no terminal outcome, or a consumer is recreated while its pooled connection remains active. Returning too few permits progressively reduces usable receiver capacity and can stall Shared delivery; returning stale or excess permits weakens backpressure.

Modifications

This documentation-only PR defines one end-to-end permit-accounting contract:

• the broker finalizes the logical-message permit count P once for every command that will actually be sent;
• an optional, backward-compatible CommandMessage.message_permits field carries P to the client;
• consumer accounting, both persistent Shared dispatcher implementations, and command serialization use the same finalized values;
• the Java native-message path treats P as a command-local budget and returns exactly that budget across delivery, skips, and supported processing failures;
• returned credit is bound to a local broker-consumer incarnation, so delayed work cannot grant old credit to a replacement consumer even when the same ClientCnx is reused; and
• an asynchronous message-write failure removes the affected broker consumer, giving unsuccessful writes an explicit terminal outcome.

The exact initial guarantee covers persistent Shared delivery and the Java native-message path. Custom MessagePayloadProcessor output, encrypted/chunked processing, non-Java clients, absolute permit reset/synchronization, and broad dispatcher refactoring remain compatible follow-up work but are intentionally out of scope.

The implementation is planned as one end-to-end PR with protocol/broker and Java-client commit layers, so the invariant and compatibility matrix can be reviewed together without mixing the code into this PIP PR.

Verifying this change

• ./gradlew quickCheck (421 actionable tasks; build successful)
• Checked Markdown structure, relative links, code fences, and trailing whitespace.

This PR changes documentation only and adds no executable runtime behavior.

Does this pull request potentially affect one of the following parts:

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

The PIP proposes one optional protobuf field with explicit presence-based fallback semantics. This PR itself only adds the design document.

[lhotari]

Thanks for writing this up — the problem is real and I verified the motivating failure against master before reviewing: on a checksum failure the Java client returns exactly one permit regardless of batch size (ConsumerImpl.java:1451 → the MessageIdData overload at :2168-2173 → discardMessage(..., 1) at :2175-2181), and broker permits are purely additive with no reconciliation anywhere (Consumer.flowPermits:905-923), so the loss is permanent. The mid-batch case returning D + K + 1 instead of P also reproduces from the code (ConsumerImpl.java:1808-1862). The P = cardinality(deliverable ack_set) model matches what the client already assumes — see the comment at ConsumerImpl.java:1813-1815. The document structure follows pip/TEMPLATE.md.

I have left seven inline comments. Three are worth settling before the vote, the rest are nits or optional:

Worth settling before the vote

1. The command budget double-returns on four existing paths (:355) — remaining = P plus "the command path returns every unit not transferred" is stated unconditionally over native command processing, but ConsumerImpl.messageReceived already returns permits itself and then returns early in four places, two of which (chunk assembly, encrypted payloads) this PIP explicitly defers. Implemented literally, those return twice.
2. The finalized-value consumer list is incomplete (:304) — finalization lands in the common Consumer.sendMessages, but four more dispatchers debit totalAvailablePermits in send loops of their own, and the sticky-key ones are @Overrides, so fixing the base loop does not reach them. Since the only production trigger for post-admission rejection other than a closing consumer is the Key_Shared draining handler, this would create a new consumer-vs-dispatcher divergence in exactly the subscription type listed as out of scope.
3. The "release the skipped message object" rule needs a dead-letter carve-out (:352) — correct for the duplicate skip, unsafe for the DLQ skip.

Nits / optional: the decompression row's stated mechanism (:152), leaving avgMessagesPerEntry undefined (:320), a question on root cause 4's motivation (:315), an optional feature flag (:411), and a link to your own in-flight #26289 (:547).

---

On preciseDispatcherFlowControl

Short answer: it does not affect the contract, but it deserves a sentence. The permit unit is logical messages regardless of the setting — neither Consumer.flowPermits (:905-923) nor the send-path debit (:433-434) consults it. It is read only in the calculateToRead implementations (PersistentDispatcherMultipleConsumers:531, ...Classic:455, PersistentDispatcherSingleActiveConsumer:450), where it converts a logical-message permit budget into an entry count to read, and it defaults to false. So P is the same either way and no rule in the PIP has to change.

The reason to mention it is indirect: its divisor Consumer.getAvgMessagesPerEntry() is maintained inside Consumer.sendMessages from precisely the counts this PIP finalizes. See the inline comment on :320 — the ask is only that the PIP say whether that EMA keeps its current inputs or moves to the finalized ones, so the "statistics do not change" sentence is unambiguous for whoever implements it. Adding preciseDispatcherFlowControl on/off to the test matrix would also be cheap insurance.

On a protocol feature flag

Line :411 is right that no flag is required, and I want to be precise about why: the invalid-value cases are already unconditional protocol errors per :331-337, and the fallback ladder at :324-329 is only reached when the field is absent. So a capability would sharpen exactly one case — absence, which :440 itself concedes is indistinguishable between an old broker and an intermediary that stripped the field. Details in the inline comment on :411; treat it as optional hardening, not a blocker.

[lhotari]

The budget rule double-returns on four paths that already return permits themselves.

remaining = P plus "the command path immediately returns all remaining units" is stated unconditionally over native command processing. But four sites inside ConsumerImpl.messageReceived already call increaseAvailablePermits and then return early — all of them before the batch branch (:1566) and before the MessagePayloadProcessor branch (:1499), so they are squarely on the native path this section governs:

Site	Today	Under the budget rule as written
ConsumerImpl.java:1474	duplicate non-batched → increaseAvailablePermits(cnx, numMessages); return;	nothing transferred, so the command path also returns P=1 → 2 returned for P=1
ConsumerImpl.java:1558	non-batched past maxRedeliverCount → increaseAvailablePermits(cnx); return;	same double return
ConsumerImpl.java:1578	non-final chunk → increaseAvailablePermits(cnx), then processMessageChunk returns null → :1512-1513 return;	double return per chunk
ConsumerImpl.java:2090	crypto DISCARD → discardMessage(..., batchSize), then :1481-1483 return;	returns B, then the command path returns P; for a complete batch P == B, so the whole batch is returned twice

That is the over-grant the PIP names as a harm at :80. Chunk assembly and encrypted payloads are deferred at :182, but they are native non-batch messages that flow through this exact path, so the deferral does not cover them.

Could this section state explicitly whether those existing increaseAvailablePermits calls are deleted as part of the change or exempted from the budget? As written an implementer reading only the PIP produces silently weakened backpressure.

Two adjacent cases worth a sentence while you are here:

• ConsumerImpl.java:1536-1543 (discard prior to startMessageId) returns no permit today — a real existing leak the budget would silently fix. Worth naming as an intended side effect so it lands with a test.
• handleCryptoFailure FAIL (:2092-2105) deliberately returns nothing and holds the message for redelivery. The budget rule changes that; say so on purpose or exempt it.

[void-ptr974]

Agreed, and thank you for identifying the four concrete paths. I addressed them through one client-ownership rule: a direct return must atomically consume the command-owned units, so later terminal cleanup returns zero. The specialized payload section now explicitly transfers authority to the existing chunk, crypto-failure, and custom-processor paths and forbids the generic budget from draining afterward. It also names the pre-startMessageId return as an intended leak fix and preserves the no-return behavior for crypto FAIL.

[lhotari]

Thanks — this is resolved, and the way you split it (one atomic terminal transition for the generic budget, explicit ownership transfer for the specialized paths) is better than the enumeration I was asking for.

Mapping my four sites onto the current text so the record is explicit:

• non-batched duplicate → :421 "Consume and return the P = 1 budget once"
• non-batched skip above the dead-letter threshold → :424 "Leave the unit command-owned"
• non-final chunk and crypto DISCARD → :485-486 "its existing permit cardinality remains authoritative and the generic budget must not also drain"
• pre-startMessageId → :407 names it as an intended correction, and :615-616 carries the test
• crypto FAIL → :486-487 keeps the no-return behavior on purpose

:404-406 is the sentence that does the work: it makes an existing increaseAvailablePermits call either budget-consuming or wrong, which is what I wanted an implementer to be unable to miss.

Two small things I'd still like your view on, then I'm happy:

1. The document never quite says that the existing increaseAvailablePermits call sites are rewritten rather than left alongside the budget. It is derivable from :404-406, but one sentence would remove the derivation step for whoever implements this.

2. "its existing permit cardinality remains authoritative" at :485 is exact for chunks, but for crypto DISCARD the existing cardinality is B, not P — discardMessage(messageId, cnx, DecryptionError, batchSize) returns batchSize. On a partial-batch redelivery where P < B that over-returns B - P, which is the harm :86 names. :178-179 does put that path out of scope, so this is not a contradiction — but since :485 now specifies the behavior rather than merely deferring it, I think it is worth saying plainly that the deferred paths may return more or less than P and that this is accepted for now. Otherwise "authoritative" reads as "exact".

Leaving this thread open for those two; nothing here changes the design.

[void-ptr974]

Thanks for the detailed review and for grounding the comments in the current implementation.

I treated the feedback as four connected gaps rather than isolated wording changes.

First, the broker now has one finalization boundary: P becomes authoritative only after admission, and the same finalized per-entry values and sum feed command serialization, consumer accounting, every applicable dispatcher aggregate, and the avgMessagesPerEntry sample. Only emitted entries and their finalized P values participate; pre-admission candidates, rejected entries, and other intermediate counts are excluded. The existing EMA behavior otherwise remains unchanged.

Second, the Java side now has one ownership model: the command budget initially owns P, each accepted native message transfers one unit to its existing lifecycle, and one terminal transition returns only the units still command-owned. Specialized payload processing and dead-letter handling have explicit ownership-transfer rules, while permit ownership remains separate from MessageImpl lifetime.

Third, I narrowed the failure scope. The decompression description now matches the current processing order, while new broker write-failure handling has been removed because the proposal did not establish a concrete gap beyond existing transport cleanup.

Finally, the compatibility tradeoffs are explicit. The PIP keeps optional-field fallback without a new capability flag, documents the ambiguity this preserves, and links the related Flow-removal work without combining its scope with this proposal.

Together, these changes preserve one end-to-end invariant: every emitted command has one finalized debit P, every covered broker counter uses that same value, and the in-scope Java lifecycle either returns exactly that debt to the source incarnation or discards it when that incarnation no longer exists.

I replied to each inline thread and left them open for your confirmation.

[lhotari]

Thanks for the very substantial revision — this is a much stronger document than the one I read. All eight points I raised are handled in the text, not just in the replies, and two of them (removing root cause 4, and the six-dispatcher aggregate table) took real work to get right.

What I checked, and what I now believe:

• The dispatcher enumeration at :336-343 matches every post-send aggregate debit in the tree today. I confirmed the six sites: PersistentDispatcherMultipleConsumers (two loops), PersistentDispatcherMultipleConsumersClassic (two loops), PersistentStickyKeyDispatcherMultipleConsumers:348, PersistentStickyKeyDispatcherMultipleConsumersClassic:401, NonPersistentDispatcherMultipleConsumers:210, and NonPersistentStickyKeyDispatcherMultipleConsumers:185. :345 ("normal, replay, and chunk-specific loops") covers the doubled sites. Nothing else under pulsar-broker/src/main debits an aggregate on send.
• The Consumer.sendMessages reading behind root cause 2 holds: the debit is MESSAGE_PERMITS_UPDATER.addAndGet(this, ackedCount - totalMessages) where totalMessages is the pre-admission parameter and ackedCount spans rejected entries too, so it over-debits by a rejected entry's P exactly as :153 says.
• Removing root cause 4 was the right call, and :184 plus :327-329 close it cleanly.

Two new points come from reading this revision, both about text the last round did not contain. Neither is a design objection — the first is a scope widening I think was unintentional, the second is a section-placement question. Details inline.

The empty mailing-list links at :685-686 are expected at this stage; worth filling before the vote thread starts.

[lhotari]

[DESIGN] the disconnect trigger widened from "invalid explicit value" to "malformed command", which now contradicts the Alternatives bullet and can fire against an old broker

In the version I reviewed, this rule read:

An invalid explicit value is a protocol error: the client sends no Flow credit and closes the source connection…

It now reads:

A malformed command returns no Flow credit and closes the source connection so the associated broker-side debt is discarded.

and the set of malformed conditions defined just above it (:375-377) grew to include several that do not involve the explicit field at all:

• B is not positive;
• a non-empty all-zero ack_set;
• an out-of-range set bit;
• native expansion beyond P.

Those are derived from the payload and from ack_set, both of which an old broker supplies. :377 says "Validation occurs before accepting a native message whenever the required metadata is available" — with no explicit-field precondition — so on an old-broker connection the client resolves P through step 2 or 3 of the ladder at :364-369 and can still reach the disconnect. "Native expansion beyond P" is the sharpest case: when P came from ack_set cardinality, exceeding it means the decoded payload disagreed with the broker's ack set, which is ordinary payload corruption.

That collides with the Alternatives bullet at :581-583, which is unchanged:

Always disconnect on ambiguous payload failure: discards the debt but disrupts every producer and consumer on a multiplexed connection. Disconnect remains appropriate for an invalid explicit value, not normal data corruption or old-protocol ambiguity.

The blast radius that bullet describes is real: ClientCnx is pooled, so closing it takes down unrelated producers and consumers, and ConsumerImpl closes the channel in only two places today — both in the subscribe path, never from messageReceived. Compare the zero-queue rule you wrote at :478, which closes the consumer; that matches ZeroQueueConsumerImpl.rejectBatchMessageByClosingConsumer, which calls closeAsync() on the consumer rather than on the channel.

I don't think you meant to widen it. Two ways to land it, either is fine:

1. keep disconnect for an invalid explicit P only, and route the metadata/ack_set inconsistencies to a bounded outcome (discard the command, return P, close the consumer) — this keeps :581-583 true as written; or
2. keep the wider rule deliberately, and update :581-583 so the two sections agree, saying explicitly that old-broker payload inconsistency can now close a shared connection.

Worth settling before the vote, since it is the one place where this PIP changes client behavior against brokers that have not been upgraded.

[lhotari]

[MINOR] the avgMessagesPerEntry change is observable in stats and in dispatcher read sizing, but appears only here and not in Public-facing Changes

This answers what I asked in the earlier thread — thank you; the zero-emitted-entries rule at :358 also closes the +Infinity edge I mentioned.

Two things I would still pin down, because "exposes that corrected value" understates the reach:

It is observable. ConsumerStatsImpl.avgMessagesPerEntry is a public stats field, surfaced through pulsar-admin topics stats and StreamingStats. Today the numerator is the pre-admission totalMessages (summed B); under this PIP it becomes summed P. For a partial-batch redelivery those differ, so an operator watching that number sees it move after upgrade. The Public-facing Changes section at :490-520 currently covers only the protocol field, and :188 puts new metrics out of scope — which is not the same thing as a changed value on an existing one. A voter reading only that section would not learn about this.

Its consumers are wider than preciseDispatcherFlowControl. The subsection heading and :613 both frame this around that setting, but getAvgMessagesPerEntry() also drives PersistentStickyKeyDispatcherMultipleConsumers:730-744, which converts maxUnackedMessages headroom into an entry estimate whenever maxUnackedMessages > 0, independent of preciseDispatcherFlowControl. Since summed P ≤ summed B, the average can only fall, so both ceil(availablePermits / avg) and estimatedRemainingPermits rise: this changes Key_Shared read sizing under redelivery-heavy load. That is arguably the correct direction, but it is a dispatch-behavior change and deserves to be named rather than inferred.

A sentence in Public-facing Changes plus a test-matrix line for the read-sizing consequence would cover both.

[lhotari]

Thanks for the write-up, and for treating the comments as connected gaps rather than line edits — that produced a better document than answering them one at a time would have.

I've confirmed each thread inline. Six are resolved; I left the budget thread open for two small follow-ups, and opened one new comment on the avgMessagesPerEntry section. Beyond those, one new point on :379: the protocol-error rule widened from "an invalid explicit value" to "a malformed command", and the malformed set at :375-377 now includes conditions derived from payload metadata and ack_set rather than from the explicit field. That makes the disconnect reachable against an old broker, which contradicts the Alternatives bullet at :581-583 that still says disconnect is not appropriate for "normal data corruption or old-protocol ambiguity". I don't think that widening was intentional; either narrowing the rule or updating the bullet would settle it.

On the verification side, I re-walked the two claims that carry the most weight for a voter. The dispatcher table at :336-343 matches every post-send aggregate debit in the tree — I found no seventh site, and :345 correctly covers the classes that have two debit loops rather than one. The root-cause-2 reading also holds against Consumer.sendMessages as it stands today.

Worth filling in the mailing-list links at :685-686 before you start the discussion thread.