[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]

This enumeration is incomplete, and the gap lands on Key_Shared.

Finalization is specified to happen in Consumer.sendMessages, which is shared by every multi-consumer dispatcher. But in both classes named here, the send loop that debits totalAvailablePermits is the overridable trySendMessagesToConsumers (PersistentDispatcherMultipleConsumers.java:780) — sendMessagesToConsumers itself is final and only delegates (:762) — and four other dispatchers have their own loop and their own debit:

• PersistentStickyKeyDispatcherMultipleConsumers.java:348-349 (an @Override at :261, so fixing the base loop does not cover it)
• PersistentStickyKeyDispatcherMultipleConsumersClassic.java:401-402
• NonPersistentDispatcherMultipleConsumers.java:210
• NonPersistentStickyKeyDispatcherMultipleConsumers.java:185

(The two persistent sticky-key sites debit -(getTotalMessages() - getTotalAckedIndexCount()); the two non-persistent ones debit the raw total. Both are computed pre-admission.)

This matters more than a missing table row. Post-admission rejection comes from PendingAcksMap.addPendingAckIfAllowed (:142-155), which returns false only when the map is closed or when the draining-hash handler rejects — and the comment at Consumer.java:400-403 says the latter is the Key_Shared case. A closing consumer is about to be removed anyway, so the one live-consumer trigger for root cause 2 is Key_Shared draining. Making Consumer.MESSAGE_PERMITS_UPDATER post-admission-accurate while leaving PersistentStickyKeyDispatcherMultipleConsumers:348-349 on the pre-admission total introduces a new consumer-vs-dispatcher divergence precisely there — while :187 lists Key_Shared draining as out of scope.

Consuming the finalized aggregate in the sticky-key dispatchers is not a change to Key_Shared routing policy, so I do not think the out-of-scope line covers it. Could the enumeration either include them or say explicitly what happens to those aggregates?

[void-ptr974]

Agreed. This is now covered by the broker finalization boundary: all six multi-consumer dispatchers consume the same finalized send sum, including both persistent sticky-key overrides and both non-persistent implementations. The rule applies to normal, replay, and chunk-specific aggregate debits without changing Key_Shared routing or draining policy.

[lhotari]

This blanket rule needs a dead-letter carve-out.

In receiveIndividualMessagesFromBatch — the method Appendix A puts in scope at :521 — only two skip branches reach a constructed MessageImpl; compacted-out, prior-batch-index and cleared-ack_set indexes all return null from newSingleMessage before an object exists. For the duplicate skip (ConsumerImpl.java:1831-1834) releasing is correct and currently leaks, which I assume is why the rule is written this way.

The other branch is the problem. The DLQ over-redelivery skip adds the message to possibleToDeadLetter at ConsumerImpl.java:1824 before skipping at :1826-1828, that list is stored into possibleSendToDeadLetterTopicMessages at :1848, and :1850-1851 then calls redeliverUnacknowledgedMessages, which reaches processPossibleToDLQ (:2355) and reads getReaderSchema(), getData(), properties, keys and event time off those same objects (:2361-2373). With poolMessages(true), MessageImpl.release() frees the payload and hands the object back to the Netty recycler (MessageImpl.java:749-754), so the DLQ producer would read a freed buffer or an unrelated recycled message.

Scoping it honestly: this needs the opt-in poolMessages(true) plus a DLQ policy, so it is not a default-configuration failure. But the rule as phrased is unconditional. Suggest narrowing it to something like "released, unless the message is retained by another component (e.g. dead-letter processing)" — or naming the duplicate skip specifically, since that is the one this is trying to fix.

The same shape exists on the non-batched path at ConsumerImpl.java:1552-1558, which stores the message for DLQ and then returns a permit.

[void-ptr974]

Agreed. The revision separates permit ownership from MessageImpl ownership. Dead-letter processing may retain the message object after the command-owned permit is returned, so a skip path must not release an object still owned by the dead-letter path. The combined duplicate/dead-letter case must either remove the object from retention before release or leave release to the retaining owner.

[lhotari]

Nit: avgMessagesPerEntry is left undefined, and it lives in the method this PIP redefines.

Consumer.sendMessages also maintains the avgMessagesPerEntry EMA (Consumer.java:423-430) from totalMessages / totalEntries. Today the numerator is the pre-admission parameter while the denominator is decremented when admission rejects an entry (Consumer.java:403), so the two already disagree. That value feeds the preciseDispatcherFlowControl read sizing (PersistentDispatcherMultipleConsumers.java:532, ...Classic:456, PersistentDispatcherSingleActiveConsumer:451, PersistentStickyKeyDispatcherMultipleConsumers:738) and is exported as the public ConsumerStats.avgMessagesPerEntry (Consumer.java:1047).

Since this PIP finalizes exactly those counts in exactly that method, could it state whether the EMA keeps its current inputs or moves to the finalized ones? That would make the "statistics do not change" sentence on this line unambiguous for the implementer.

One edge case worth pinning down either way: if admission rejects every surviving entry, totalEntries reaches 0 while totalMessages > 0, so :426 evaluates 1.0 * totalMessages / 0 and the EMA becomes +Infinity permanently. getAvgMessagesPerEntry() (:961-962) is (int) Math.round(+Inf) = (int) Long.MAX_VALUE = -1. Every consumer clamps it with Math.max(1, ...), so read sizing degrades rather than breaks, but pulsar-admin topics stats reports -1 until it is reset from stats at :1032. Pre-existing, but this is the natural place to fix or explicitly leave it.

[void-ptr974]

Agreed. The existing avgMessagesPerEntry EMA remains unchanged. Its input sample is derived only from the finalized send result:

sum(finalized P values for emitted entries) / number of emitted entries.

Pre-admission candidates, rejected entries, and other intermediate counts do not participate. When no entry is emitted, the EMA is not updated. The permit unit remains unchanged with preciseDispatcherFlowControl enabled or disabled, and the test matrix covers both settings.

[lhotari]

Nit (documentation accuracy): the mechanism clause is right for checksum and metadata, but not for decompression.

messageReceived parses metadata and binds B at ConsumerImpl.java:1465 (final int numMessages = msgMetadata.getNumMessagesInBatch();), and only decompresses afterwards at :1491-1492; uncompressPayloadIfNeeded even takes the parsed MessageMetadata as a parameter. So on decompression failure B is known and simply ignored — :2123/:2135 call the MessageIdData overload of discardCorruptedMessage, which hardcodes discardMessage(..., 1) (:2168-2181).

The Permit-consequence column is correct either way, and root cause 1 in the body already gets this right. Just worth splitting the row (or the clause) so the "before B is available" claim only covers checksum and metadata — for decompression the existing B would suffice, which slightly narrows the case for a wire field.

[void-ptr974]

Agreed. I corrected the failure classification: B is unavailable for checksum and metadata failures, but is already available for decompression failure and is currently ignored by the generic corrupted-message path.

[lhotari]

Question on root cause 4's motivation. The literal statement at :139-141 is accurate — the debit does precede the write (Consumer.java:434 vs :441) and the listener does only log (:454-458). But "Leaving that consumer live strands both the undelivered permit debt and any pending-ack ownership" looks overstated, which weakens the case for this remedy.

Per-message CommandMessage writes are issued with ctx.voidPromise() (PulsarCommandSenderImpl.java:297-301); only the trailing Unpooled.EMPTY_BUFFER carries the real writePromise (:306). A void promise's failure fires exceptionCaught, and ServerCnx.exceptionCaught ends in ctx.close() (:628), after which channelInactive closes every registered consumer on that connection (:467-485). So for the transport failures that dominate in practice, the consumer is already removed by the existing path.

There is also a circularity in the remedy itself: it routes through the consumer-disconnect path, which writes CommandCloseConsumer on the same channel whose write just failed and closes the physical connection when that write fails too (ServerCnx.java:4295-4315) — i.e. it converges on what channelInactive already does.

I am not claiming the rule is useless: a non-IOException write error half-closes rather than closes, which may be exactly the window you have in mind. Could you name the concrete failure mode where the write promise fails, the channel stays open, and the Consumer stays registered? That would justify the rule and give the test at :498 something specific to assert.

[void-ptr974]

Agreed. I could not establish a sufficiently concrete case requiring new behavior beyond the existing transport cleanup. I therefore removed broker write failure as a root cause, removed the proposed disconnect behavior and test requirement, and made new broker write-failure handling explicitly out of scope.

[lhotari]

Optional: a FeatureFlags capability would sharpen exactly one case.

To be precise about what a flag would and would not buy, since this line is otherwise correct that none is required:

• Zero, out-of-range and metadata-inconsistent values are already unconditional protocol errors per :331-337, so a flag adds nothing there.
• The fallback ladder at :324-329 is reached only when the field is absent — and :440 concedes that absence is indistinguishable between an old broker, an upgraded broker that missed a serialization path, and an intermediary that stripped the field. That is the one case a capability makes decidable, which matters because the silent outcome there is the legacy one-permit path, i.e. the exact leak this PIP exists to remove. It would also give the Monitoring section (:415-420) the signal it currently says it lacks.

If you want it, the machinery is already live in both directions and the next free tag is 10:

optional bool supports_message_permits = 10 [default = false];

advertised broker→client in CommandConnected.feature_flags only when every debited CommandMessage on that connection carries a valid value; on a flagged connection a missing field becomes a protocol error rather than a silent fallback; on an unflagged connection the ladder is unchanged. No client→broker capability is needed, since the broker emits the field unconditionally and old clients ignore unknown optional fields.

Two implementation details worth recording if you adopt it:

• Commands.newConnectedCommand takes an explicit boolean per flag (Commands.java:320-344), so its signature and both callers change.
• The proxy does not forward CommandConnected — ProxyConnection.handleBrokerConnected regenerates it and copies only a hand-picked subset of flags (ProxyConnection.java:515-527), so a new flag is dropped through the Pulsar proxy without explicit plumbing. The per-command field itself is proxy-safe: post-handshake frames are forwarded verbatim (DirectProxyHandler.java:300-317), which the PIP correctly states at :444-446.

Non-blocking either way — a version bump would work too (Commands.java:331-336 advertises min(server, client)), but recent capability additions have all used FeatureFlags.

[void-ptr974]

Thanks for clarifying exactly what the capability would distinguish. I kept it as an alternative rather than expanding this PIP with handshake and proxy capability plumbing. The compatibility and monitoring sections now state that absence cannot distinguish an old broker, an omitted serialization path, or field stripping, and the test matrix includes a field-stripping intermediary.

[lhotari]

Nit: :191 says "Known adjacent work is listed in Appendix B", but Appendix B lists categories without links. Its first bullet — asynchronous CommandFlow dispatcher updates racing with consumer removal — already has an open fix of yours in flight, #26289 (and issue #26288), labelled release/4.2.5 and release/4.0.14, which touches Consumer's permit counters and totalAvailablePermits in both Shared dispatchers.

Linking it here would save voters the search and flag the overlap for whoever implements this — the two changes land in the same methods, so merge order is worth stating.

[void-ptr974]

Added #26288 and #26289 as related work. The document now records the overlapping counters and merge-order consideration while keeping the Flow-removal race separate from this PIP send-side accounting scope.

[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.