From ffb580bad9e308ef0146aa887848eb4b8533c017 Mon Sep 17 00:00:00 2001 From: void-ptr974 Date: Sat, 15 Aug 2026 17:41:44 +0800 Subject: [PATCH 1/7] [improve][pip] PIP-491: Explicit permit accounting for batched message delivery Assisted-by: Codex (GPT-5) --- pip/pip-491.md | 471 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 471 insertions(+) create mode 100644 pip/pip-491.md diff --git a/pip/pip-491.md b/pip/pip-491.md new file mode 100644 index 0000000000000..e990307f384e8 --- /dev/null +++ b/pip/pip-491.md @@ -0,0 +1,471 @@ +# PIP-491: Explicit permit accounting for batched message delivery + +# Background knowledge + +Pulsar consumers use permits for receiver-side flow control. A client sends `CommandFlow` to grant receiver +capacity, the broker consumes that capacity while dispatching, and the client returns permits as messages leave its +prefetch path. In the consumer flow accounting covered by this PIP, one permit represents one logical message, not +one BookKeeper entry. + +A native Pulsar batch stores multiple logical messages in one entry. The payload metadata field +`num_messages_in_batch` describes the original batch cardinality so the broker can account for the entry and the +client can deserialize it. This PIP calls that value `B`; a non-batched entry has `B = 1`. + +With batch-index acknowledgments introduced by [PIP-54](pip-54.md), a redelivered entry can include an `ack_set` in +`CommandMessage`. In this representation, a set bit identifies an unacknowledged index that remains deliverable; a +cleared bit identifies an index that the client must not deliver. A complete batch therefore consumes `B` permits, +while a partial batch consumes only the number of deliverable indexes. + +This PIP calls the actual permit debit for one `CommandMessage` `P`. For a valid native batch: + +```text +complete batch: P = B +partial batch: P = cardinality(deliverable ack_set indexes within B) +valid command: 1 <= P <= B +``` + +For example: + +| Entry | `B` | Deliverable `ack_set` indexes | `P` | +| --- | ---: | --- | ---: | +| Non-batched message | 1 | absent | 1 | +| Complete native batch | 8 | absent | 8 | +| Partial native batch | 8 | `{0, 2, 5}` | 3 | + +The partial example shows why `B` cannot be reused as the debit: the payload still contains eight original indexes, +but only three logical messages are eligible for this delivery. + +The acknowledgment path is separate from permit accounting. Acknowledgment controls cursor progress and redelivery; +permits control subsequent dispatch. Acknowledging an entry does not repair a missing permit, and returning a permit +does not acknowledge a message. + +Permit accounting is also connection-scoped. When a client consumer reconnects, the broker replaces its server-side +consumer. Debt belonging to the removed consumer must not be compensated on the new connection. + +# Motivation + +For each sent `CommandMessage`, the broker must debit `P`, but the command does not carry `P`. The Java client must +infer it from `ack_set` and the payload. If the payload cannot be parsed, the client may not know `P`; if only part of +a batch enters the message lifecycle, the current code does not preserve how ownership of `P` was split. + +The current protocol therefore relies on duplicated computation across broker layers and the client. The dispatcher, +consumer, and sender derive related counts at different points in the send lifecycle, while the client independently +reconstructs the expected return from what it can still observe after receipt. Equality of those calculations is an +implementation coincidence, not an explicit wire contract. It breaks when those paths observe different information, +such as post-selection admission changing the actual send set or a payload failure before the client can read batch +metadata. + +This PIP does not add periodic synchronization, a permit reset, or a per-command acknowledgment. It establishes a +per-command accounting contract instead: the broker is the authority that creates and declares a debt of `P`, and +the Java client must conserve and eventually return exactly `P` while the source connection remains active. If that +connection is replaced, its broker consumer and debt disappear together, so the client discards the old credit. +`CommandFlow` can continue to aggregate returned credit; it no longer needs the client to independently rediscover +how the debt was created. + +Steady-state delivery does not normally lose permits: the duplicated calculations agree for successful complete, +partial, and non-batched delivery. They diverge when the actual send set changes after an earlier aggregate was +calculated, payload processing fails before or during batch expansion, a handed-off message has no terminal outcome, +or the connection changes while credit is being returned. + +For example, assume a Shared consumer grants 100 permits and the broker sends one native batch with `P = 100`. If +checksum verification fails before the client can parse `num_messages_in_batch`, the current Java error path returns +one permit. The broker consumer is left with one available permit instead of 100. Repeating the failure can exhaust +its capacity and stall Shared dispatch. + +```mermaid +sequenceDiagram + participant D as Shared dispatcher + participant B as Broker consumer + participant J as Java consumer + + D->>B: Select native batch, P = 100 + B->>B: Debit 100 permits + B->>J: CommandMessage plus payload, without P + J->>J: Checksum fails before B can be parsed + J-->>B: Return 1 permit + Note over B,J: 99 permits remain unmatched +``` + +There are three root causes. + +## 1. The wire protocol relies on duplicated inference instead of an explicit debit + +`num_messages_in_batch` cannot be the permit contract because it is inside the payload metadata. It is unavailable +when checksum or metadata parsing fails, and it remains `B` when a partial redelivery has `P < B`. + +`ack_set` is sufficient for a partial batch but is normally absent for a complete batch. Neither existing field is a +reliable statement of the debit for every command. + +## 2. The broker does not finalize one value and pass it end to end + +The dispatcher builds batch and aggregate counts before final send eligibility is known. `Consumer.sendMessages` +then records pending-ack ownership for individual-ack subscriptions; this final admission step can reject an entry, +for example when the consumer has closed. The Shared dispatcher and command sender consume related state through +separate paths. Even when their formulas agree on the normal path, there is no finalized per-command value shared by +consumer accounting, dispatcher accounting, and serialization. + +For example, if two entries each contain ten logical messages and final admission rejects the second entry, only one +command is sent. Any calculation that still uses the earlier aggregate of 20 disagrees with the actual send of ten. +A wire field is useful only if it is populated from the post-admission value. + +## 3. The Java client has no explicit command-to-message ownership boundary + +The native batch path separately counts delivered, skipped, and failed work. It does not retain one value representing +the part of `P` still owned by command processing. Once delivery is handed to an asynchronous callback, terminal +handling and source-connection association are also spread across separate paths. + +For `P = 10`, if four messages enter the normal lifecycle before a later index fails, those messages should own four +units and command processing should retain six. The current delivered, skipped, and generic error counts do not +represent that split directly. + +Adding `P` only to corrupted-message helpers would fix some early failures, but would not define who returns each unit +during partial deserialization, queue rejection, or reconnect. The client therefore needs a command-local budget as +well as the wire value. + +## Current failure modes and evidence + +| Failure mode | Evidence in the current path | Permit consequence | +| --- | --- | --- | +| Checksum, metadata, or decompression failure | Direct: `ConsumerImpl.messageReceived` reaches corrupted-message handling that returns one. | A command with `P > 1` loses `P - 1` permits. | +| Failure partway through native batch parsing | Direct: `ConsumerImpl.receiveIndividualMessagesFromBatch` lets already-transferred messages return individually, counts client-side skips separately, and returns one from the catch path. | The total is not tied to `P` and can be too small or too large. | +| Local queue offer rejects an asynchronously handed-off message | Direct: `ConsumerBase.enqueueMessageAndCheckBatchReceive` rolls back queue-size accounting, but has no matching terminal handling for the message and its permit. | One debited unit can be stranded. | +| Connection replacement during asynchronous handling | Race-dependent: the source `ClientCnx` is available at command receipt, but later native-batch and callback paths can read the consumer's current `cnx()`. | Credit belonging to the old broker consumer can be dropped or applied to its replacement. | +| Shared pending-ack admission rejects a batched entry after dispatch selection | Race-dependent: `Consumer.sendMessages` can null and release an entry when its `PendingAcksMap` has closed, while later consumer and dispatcher debits still originate from the earlier aggregates and the sender skips the null entry. | The server can debit permits for a command that is not sent. | + +For the partial-parsing case, suppose `D` messages were transferred and `K` deliverable messages were skipped by the +client before the failure. The current paths eventually return `D + K + 1`, regardless of how many deliverable +indexes remain. This under-returns when more than one remains and can over-return when none remains but parsing later +fails on an index that was not part of `P`. + +The first three rows follow directly from current terminal branches; the connection and closed pending-ack cases need +deterministic race-oriented tests. The table describes native-batch delivery, including common queue and connection +helpers only where that path reaches them. The asynchronous carrier lifetime in Appendix A is an implementation risk, +not a separately claimed production failure. + +# Goals + +## In Scope + +- Add the actual logical-message debit `P` to `CommandMessage`. +- Finalize `P` once on the broker after send eligibility is known. +- Reuse that result for consumer accounting, persistent Shared dispatcher accounting, and command serialization. +- Conserve `P` in the Java native-batch path across successful delivery, client-side skips, and checksum, metadata, + decompression, and batch-deserialization failures. +- Bind immediate and deferred permit returns to the source client connection. +- Preserve old/new broker, Java client, and Pulsar proxy interoperability. +- Validate the result end to end with persistent Shared subscriptions. + +## Out of Scope + +- Custom `MessagePayloadProcessor` output accounting. +- Encrypted or chunked message processing. +- Non-Java client implementations. +- Permit reset, absolute synchronization, or connection-generation commands. +- General Flow/consumer-removal races and Key_Shared-specific draining behavior. +- Changes to acknowledgment, redelivery, public receiver-queue, dispatch-rate, or message-rate semantics. +- New Java public APIs, configuration, CLI commands, metrics, or broad consumer/dispatcher refactoring. + +Known adjacent problems are listed in Appendix B to make the boundary explicit. + +# High Level Design + +For a send containing one or more entries, the broker produces one finalized `P` for every entry that will actually +be sent. It also calculates the sum of those per-command values. The per-command values and their sum become the only +inputs to the permit-related server counters and command serialization covered by this PIP. + +The Java client treats `P` as a command-local budget. Each native message accepted into the existing per-message +lifecycle takes one unit. That lifecycle later returns the unit. The command path returns every unit not transferred +to a message. The same finalized broker value and the ownership split are shown below. + +```mermaid +flowchart TB + subgraph Broker + A[Surviving entries] --> F[Finalize one P per command] + F --> C[Debit Consumer by sum P] + F --> S[Debit persistent Shared dispatcher by sum P] + F --> W[Serialize P in each CommandMessage] + end + + subgraph Java + W --> R[Resolve and validate P] + R --> Q[Command-owned budget P] + Q -->|D units claimed| M[D native messages] + Q -->|P minus D units remain| T[Command completion or failure] + M -->|one per terminal message| X[Return D permits to source connection] + T --> Y[Return P minus D permits to source connection] + end +``` + +For every successfully delivered send on a live consumer connection, the invariant is: + +```text +broker consumer debit == persistent Shared dispatcher debit == sum of all P values in the send + +for each command: + CommandMessage.message_permits == Java client's eventual return == P +``` + +Non-batched delivery naturally uses `P = 1` and needs no separate accounting model. + +## How the design closes the current failures + +The protocol field is necessary but is not sufficient by itself. The guarantee comes from applying the complete +broker and Java ownership rules together: + +| In-scope scenario | Preventing mechanism | Result with a new broker and new Java client | +| --- | --- | --- | +| Checksum, metadata parsing, or decompression fails before batch expansion | `P` is present in `CommandMessage` and resolved before payload processing. | The command path returns `P` without expanding the batch. | +| Native batch parsing fails after `D` messages are handed off | The command-local budget transfers only `D` units and retains `P - D`. | Messages return `D`; command failure returns `P - D`; total return is `P`. | +| A client-side skip occurs before message handoff | The skipped index never claims a unit from the command budget. | Its unit remains in `P - D` and is returned by command completion. | +| Local queue offer rejects a handed-off message | Queue rejection becomes a terminal message path that releases the message and returns its claimed unit. | The unit is neither leaked nor returned twice. | +| The source connection is replaced during asynchronous handling | The command budget and transferred messages retain the source connection. | Old credit is discarded and is never applied to the replacement consumer. | +| Pending-ack admission rejects an entry selected earlier | Broker finalizes per-entry values after admission and derives every covered debit from their sum. | The rejected entry produces no command and no permit debit. | + +The full in-scope guarantee requires a new broker and new Java client; mixed-version behavior is defined below. + +# Detailed Design + +## Design & Implementation Details + +### Broker permit finalization + +Logically, finalization occurs in `Consumer.sendMessages`, after pending-ack admission and before handing entries to +the asynchronous command sender. At that point the broker knows which entries will actually produce commands: + +```text +entry is not sent => no command and no debit +ack_set is absent => P = B +ack_set is present => P = deliverable set-bit count within B +``` + +An entry with `P = 0` is not sent. A serialized `CommandMessage` always represents at least one logical message. + +Finalization produces the per-entry `P` values and their sum. The same result is used for: + +- the consumer's available-permit debit; +- the consumer's unacked-message accounting where applicable; +- the persistent Shared dispatcher's aggregate available-permit debit; and +- each command's `message_permits` field. + +Downstream layers must consume this result rather than recompute it from `B`, `ack_set`, or aggregate message counts. +Finalization is synchronous even though the network write is asynchronous: the Shared dispatcher must capture the +finalized sum as part of the send handoff, before the sender can recycle its input. The per-entry values remain valid +until the sender has serialized them. The carrier type, allocation strategy, and method signatures are implementation +details discussed in Appendix A. + +This PIP does not change dispatch-rate, byte-rate, or public message-rate statistics. + +### Java permit resolution + +For the native path, the Java client resolves `P` in this order: + +1. Use a present `message_permits` value. +2. Otherwise, use the cardinality of a non-empty `ack_set`. +3. Otherwise, after parsing metadata, use `num_messages_in_batch`. +4. If an old-broker command fails before either source is available, use the legacy value of one. + +A present `message_permits` value is authoritative for flow accounting once validated. It must be in the range +`1..Integer.MAX_VALUE`; Java must interpret and validate the protobuf `uint32` value using unsigned conversion. + +After native metadata is available, an explicit value must equal `B` for a complete batch or the deliverable +`ack_set` cardinality for a partial batch. This validation happens before any native message is transferred to the +per-message lifecycle. An invalid explicit value is a protocol error: the client sends no Flow credit and closes the +source connection so the associated broker-side debt is discarded. + +The final fallback cannot recover a complete-batch debit from an old broker when failure occurs before metadata +parsing. Exact conservation for that case requires an upgraded broker. The client retains the legacy one-permit +behavior rather than closing a multiplexed connection for this old-protocol ambiguity. + +### Java native-batch budget + +Native command processing starts with `remaining = P`. + +Indexes excluded by the broker's `ack_set` are not part of `P` and never interact with the budget. For each +deliverable index, immediately before its logical message is accepted into the existing per-message lifecycle, the +client claims one unit by decrementing `remaining`. If the transfer itself fails, the claim is restored. + +A deliverable index that the client later skips as duplicate, compacted-out, or otherwise ineligible does not claim a +unit, so its unit remains in the command budget. Any message object already created for that skipped index is +released. + +At normal completion or failure, the command path immediately returns all remaining units. If deserialization fails +after `D` messages were transferred, those messages eventually return `D`, while the command path returns `P - D`. + +Processing more than `P` native messages is a malformed command and follows the protocol-error behavior above. + +Once a message claims one unit, ownership passes to the existing per-message lifecycle. Every terminal branch in that +lifecycle must eventually return the unit, either after normal prefetch processing or while releasing a rejected +message. The required Shared-path terminal fix includes local queue-offer rejection; it does not redesign +receiver-queue behavior or add an arbitrary permit cost to `Message`. + +The budget and every message created from it retain the `ClientCnx` that delivered the command. Credit is accumulated +or sent only while that source connection is still active for the consumer. If it has been replaced, the credit is +discarded because the corresponding broker-side consumer and its debt have already been removed. Old-connection +credit is never applied to the replacement consumer. + +## Public-facing Changes + +### Binary protocol + +Add one optional field to `CommandMessage`: + +```proto +message CommandMessage { + required uint64 consumer_id = 1; + required MessageIdData message_id = 2; + optional uint32 redelivery_count = 3 [default = 0]; + repeated int64 ack_set = 4; + optional uint64 consumer_epoch = 5; + + // Number of consumer permits debited by the broker for this command. + optional uint32 message_permits = 6 [default = 1]; +} +``` + +The wire rules are: + +- valid values are `1..Integer.MAX_VALUE`; +- the broker may omit the field when `P = 1`, and sends it when `P > 1`; +- zero is invalid; no command is sent when nothing is deliverable; and +- the value is the actual debit, not necessarily the original batch size. + +No protocol-version bump or feature flag is required. Old protobuf readers ignore the optional field, so an upgraded +broker can send it without detecting client support. New Java clients use field presence and the fallback rules. + +# Monitoring + +No metric is added. Existing consumer `availablePermits`, receive-failure statistics, and dispatch progress can help +diagnose the behavior. Tests verify the accounting invariant directly and verify that repeated corrupted batches do +not stall a Shared consumer. + +# Security Considerations + +The broker-to-client field grants no client authority and does not change authentication, authorization, or tenant +isolation. +The Java client must reject zero and unsigned values above `Integer.MAX_VALUE` and must not return more than `P` for +one command. Permit accumulation uses checked arithmetic; overflow is handled as a protocol error on the source +connection rather than wrapping into negative or unrelated Flow credit. + +# Backward & Forward Compatibility + +| Broker | Java client | Behavior | +| --- | --- | --- | +| New | New | Exact native-batch accounting from explicit `P`. | +| New | Old | The field is ignored; normal delivery is unchanged, while old exceptional-path limitations remain. | +| Old | New | The client falls back to `ack_set`, parsed batch metadata, then legacy one. | +| Old | Old | No change. | + +Omitting `P = 1` is unambiguous. A complete one-message command falls back to one, while a partial batch with one +deliverable index carries an `ack_set` with cardinality one. An old client already returns one permit per successfully +delivered native message, so an upgraded broker does not change its normal path. + +Pulsar proxy paths forward the original message frame without re-serializing `CommandMessage`. A third-party proxy +that drops unknown fields remains wire-compatible through the fallback, but cannot preserve the new early-failure +guarantee for complete batches. + +## Upgrade + +No configuration or metadata migration is required. Upgrade brokers before Java clients when practical. Old clients +can remain connected to upgraded brokers. + +## Downgrade / Rollback + +The value is connection-local and not persisted. Brokers can be rolled back without data migration; new Java clients +then use the fallback rules. + +## Pulsar Geo-Replication Upgrade & Downgrade/Rollback Considerations + +`CommandMessage` is not stored or replicated between clusters, so geo-replication requires no coordinated upgrade or +rollback. + +# Alternatives + +- **Continue inferring from batch metadata:** cannot handle checksum or metadata failure before `B` is parsed and + cannot describe a partial-batch debit by itself. +- **Use only `ack_set`:** exact for partial batches, but complete batches generally omit it. +- **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. +- **Add a reset or absolute-permit command:** requires ordering, idempotency, in-flight command, and connection- + generation rules. Attaching the debit to the command that creates it is smaller. +- **Assign arbitrary permit costs to `Message`:** unnecessary for native batches, where each accepted logical message + owns exactly one unit. Custom processors need a separate entry-level completion design. +- **Separate broker and client PIPs:** risks defining different debit and return semantics. Implementation can still + be staged under one end-to-end protocol contract. + +# General Notes + +After the PIP is accepted, implementation is split into two focused pull requests: + +1. **Protocol and broker:** add the field, finalize one per-entry value, align consumer and persistent Shared + aggregate debits, serialize that value, and test old-reader compatibility. +2. **Java native batch:** implement resolution and fallbacks, command-local ownership, source-connection binding, the + narrow terminal fixes, and the persistent Shared end-to-end test. + +The first PR does not introduce client lifecycle abstractions. The second PR does not redesign dispatcher selection +or Flow/removal ordering. Neither PR alone establishes the end-to-end guarantee. + +The required test matrix covers: + +- full and partial native batches, including `P = 1` omission; +- filtering or pending-ack rejection before broker finalization; +- consumer and persistent Shared debits equal the sum of the per-command values, while each command carries its `P`; +- asynchronous ownership and recycling of the broker's per-entry carrier; +- explicit-field validation and old-broker `ack_set`/metadata fallbacks; +- checksum, metadata, decompression, and mid-batch deserialization failures; +- native skips, queue rejection, and source-connection replacement; and +- repeated corrupted batches on a persistent Shared subscription without dispatch stall. + +# Appendix A: Current implementation map + +This appendix records where the current behavior is distributed. It is supporting evidence, not an additional +implementation scope. + +| Component | Current responsibility | Relevance to this PIP | +| --- | --- | --- | +| `MessageMetadata.num_messages_in_batch` | Stores original payload cardinality `B`. | Payload structure is not always the actual debit and is unavailable before metadata parsing. | +| `AbstractBaseDispatcher.filterEntriesForConsumer` | Extracts `B`, `ack_set`, and aggregate counts. | Runs before final pending-ack admission. | +| `Consumer.sendMessages` | Performs pending-ack admission and debits consumer/unacked counts. | Must finalize the values for entries that survive admission. | +| `PersistentDispatcherMultipleConsumers` | Debits aggregate `totalAvailablePermits`. | Must consume the same finalized aggregate rather than recalculate it. | +| `PulsarCommandSenderImpl.sendMessagesToConsumer` | Serializes surviving entries and `ack_set`. | Must serialize the finalized per-entry value without recomputation. | +| `EntryBatchIndexesAcks` | Carries mutable pooled per-entry `ack_set` state. | The sender owns and recycles it asynchronously, so finalized values need an explicit lifetime and must not be recomputed after handoff. | +| `ConsumerImpl.messageReceived` | Verifies checksum, parses metadata, decompresses, and chooses the payload path. | Must resolve `P` before payload-dependent failures. | +| `ConsumerImpl.receiveIndividualMessagesFromBatch` | Splits native batches and counts skipped messages. | Must transfer units from the command budget instead of reconstructing returns. | +| `ConsumerImpl.executeNotifyCallback` | Performs asynchronous callback handling. | A transferred unit needs an exact terminal outcome on its source connection. | +| `ConsumerBase.enqueueMessageAndCheckBatchReceive` | Admits messages to the local queue. | Queue-offer rejection must release the native message and its unit. | + +The broker carrier introduced by this PIP should continue the existing low-allocation approach: use primitive pooled +storage where practical and avoid allocating one object per dispatched entry. This performance constraint does not +change the protocol contract. + +# Appendix B: Related work outside this PIP + +The following known areas have different ordering or ownership requirements and are deliberately deferred: + +- asynchronous `CommandFlow` dispatcher updates racing with consumer removal; +- Key_Shared hash draining and admission policy; +- custom `MessagePayloadProcessor` output cardinality; +- encrypted payload and chunk-assembly lifecycles; +- non-Java client prefetch implementations; +- multi-topic parent and zero-queue specializations beyond behavior reached through the native child consumer; +- channel write failure and teardown for commands that were not successfully delivered; and +- broader entry, batch, and logical-message unit consistency in statistics and rate limiting. + +These areas may reuse the explicit field later, but they do not block the native Java Shared invariant. + +# Links + +* Related broader permit-unit accounting issue (not the specific failure addressed here): + https://github.com/apache/pulsar/issues/23263 +* Batch-index acknowledgment: [PIP-54](pip-54.md) +* Current protocol: [PulsarApi.proto](../pulsar-common/src/main/proto/PulsarApi.proto) +* Current broker paths: + [AbstractBaseDispatcher](../pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java), + [Consumer](../pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java), + [persistent Shared dispatcher](../pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java), + and [command sender](../pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSenderImpl.java) +* Current Java paths: + [ConsumerImpl](../pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java) + and [ConsumerBase](../pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java) +* Mailing List discussion thread: +* Mailing List voting thread: From a8bca32a0746c2d9c9b6273f4a4cf095817fc01a Mon Sep 17 00:00:00 2001 From: void-ptr974 Date: Sat, 15 Aug 2026 18:16:17 +0800 Subject: [PATCH 2/7] [improve][pip] Clarify broker and Java client responsibilities Assisted-by: Codex (GPT-5) --- pip/pip-491.md | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/pip/pip-491.md b/pip/pip-491.md index e990307f384e8..658bae3100e8a 100644 --- a/pip/pip-491.md +++ b/pip/pip-491.md @@ -207,6 +207,25 @@ for each command: Non-batched delivery naturally uses `P = 1` and needs no separate accounting model. +## Broker and Java client responsibilities + +The accounting contract has one authority and one owner at each stage. The broker creates the debt, the protocol +transfers its value, and the Java client conserves the debt until it is returned or its source connection disappears. + +| Component | Responsibility | +| --- | --- | +| Broker admission | Decide which entries will produce commands. A rejected entry produces neither a command nor a permit debit. | +| Broker finalization | Calculate one `P` for every admitted command and its sum exactly once. All covered broker counters and serialization consume that result without recomputing it. | +| Wire protocol | Carry the broker's actual per-command debit in `CommandMessage.message_permits`; this is a declaration of existing debt, not a client request. | +| Java command path | Resolve and validate `P`, create `remaining = P`, transfer one unit to each accepted native message, and return every unit that was not transferred. | +| Java message lifecycle | Return exactly one transferred unit on every terminal path, including normal consumption and local rejection. | +| Connection boundary | Associate the debit and all returns with the connection that delivered the command. Connection teardown discards that connection's debt and credit; neither is moved to a replacement consumer. | +| Mixed versions | Use the explicit value when present and valid; otherwise retain the defined `ack_set`, metadata, and legacy-one fallbacks. | + +The complete invariant is provided only by the broker and Java responsibilities together. The protocol field alone +does not fix post-admission broker recounting or client lifecycle leaks, and either implementation PR alone preserves +mixed-version behavior without claiming the new end-to-end guarantee. + ## How the design closes the current failures The protocol field is necessary but is not sufficient by itself. The guarantee comes from applying the complete @@ -247,12 +266,20 @@ Finalization produces the per-entry `P` values and their sum. The same result is - the persistent Shared dispatcher's aggregate available-permit debit; and - each command's `message_permits` field. +Unacked-message accounting and permit accounting remain separate mechanisms: unacked state controls acknowledgment +backpressure, while permits control dispatch capacity. They reuse the same finalized logical-message count only so +that neither path describes messages that final admission rejected. + Downstream layers must consume this result rather than recompute it from `B`, `ack_set`, or aggregate message counts. Finalization is synchronous even though the network write is asynchronous: the Shared dispatcher must capture the finalized sum as part of the send handoff, before the sender can recycle its input. The per-entry values remain valid until the sender has serialized them. The carrier type, allocation strategy, and method signatures are implementation details discussed in Appendix A. +Once the send is handed to the network, the debit belongs to that source connection. An asynchronous write failure is +resolved by connection teardown, which removes the associated broker consumer and its debt; the broker does not +re-credit `P` to a replacement consumer or attempt to compensate it with a later `CommandFlow`. + This PIP does not change dispatch-rate, byte-rate, or public message-rate statistics. ### Java permit resolution From 5d786b13dc523e68700dfb8f947e7905a3cd6d41 Mon Sep 17 00:00:00 2001 From: void-ptr974 Date: Sun, 16 Aug 2026 13:31:58 +0800 Subject: [PATCH 3/7] [improve][pip] Finalize PIP-491 permit accounting contract --- pip/pip-491.md | 309 +++++++++++++++++++++++++++++-------------------- 1 file changed, 186 insertions(+), 123 deletions(-) diff --git a/pip/pip-491.md b/pip/pip-491.md index 658bae3100e8a..06ad80629f58c 100644 --- a/pip/pip-491.md +++ b/pip/pip-491.md @@ -7,6 +7,12 @@ capacity, the broker consumes that capacity while dispatching, and the client re prefetch path. In the consumer flow accounting covered by this PIP, one permit represents one logical message, not one BookKeeper entry. +The broker represents this capacity at two levels for persistent Shared subscriptions. Each `Consumer` tracks its +own available permits, while the Shared dispatcher tracks aggregate `totalAvailablePermits` across its consumers to +decide whether dispatch can continue. A send must therefore debit both levels by the same number of logical messages. +If they use different counts, the dispatcher and the selected consumer disagree about remaining capacity even when +the network write itself succeeds. + A native Pulsar batch stores multiple logical messages in one entry. The payload metadata field `num_messages_in_batch` describes the original batch cardinality so the broker can account for the entry and the client can deserialize it. This PIP calls that value `B`; a non-batched entry has `B = 1`. @@ -39,8 +45,9 @@ The acknowledgment path is separate from permit accounting. Acknowledgment contr permits control subsequent dispatch. Acknowledging an entry does not repair a missing permit, and returning a permit does not acknowledge a message. -Permit accounting is also connection-scoped. When a client consumer reconnects, the broker replaces its server-side -consumer. Debt belonging to the removed consumer must not be compensated on the new connection. +Permit accounting is scoped to a broker-consumer incarnation, not merely to a physical connection. A broker can close +and recreate one consumer while the pooled `ClientCnx` remains active for other producers or consumers. Debt belonging +to the removed broker consumer must not be compensated to its replacement, even when both use the same `ClientCnx`. # Motivation @@ -57,15 +64,21 @@ metadata. This PIP does not add periodic synchronization, a permit reset, or a per-command acknowledgment. It establishes a per-command accounting contract instead: the broker is the authority that creates and declares a debt of `P`, and -the Java client must conserve and eventually return exactly `P` while the source connection remains active. If that -connection is replaced, its broker consumer and debt disappear together, so the client discards the old credit. +the in-scope Java native-message path must conserve and eventually return exactly `P` while the source +broker-consumer incarnation remains active. When that incarnation is removed, its debt disappears with it, so the +client discards old credit rather than moving it to a replacement. `CommandFlow` can continue to aggregate returned credit; it no longer needs the client to independently rediscover how the debt was created. -Steady-state delivery does not normally lose permits: the duplicated calculations agree for successful complete, -partial, and non-batched delivery. They diverge when the actual send set changes after an earlier aggregate was +Steady-state native delivery does not normally lose permits: the duplicated calculations agree for successful +complete, partial, and non-batched delivery. They diverge when the actual send set changes after an earlier aggregate was calculated, payload processing fails before or during batch expansion, a handed-off message has no terminal outcome, -or the connection changes while credit is being returned. +or the broker-consumer incarnation changes while credit is being returned. + +Returning too little permanently reduces the consumer's effective capacity and can eventually stall dispatch. +Returning too much weakens receiver-side backpressure by granting capacity the client did not originally provide. +Returning old credit to a replacement consumer crosses the incarnation boundary and can produce the same +over-grant even when the numeric total on the old consumer would otherwise have been correct. For example, assume a Shared consumer grants 100 permits and the broker sends one native batch with `P = 100`. If checksum verification fails before the client can parse `num_messages_in_batch`, the current Java error path returns @@ -86,7 +99,7 @@ sequenceDiagram Note over B,J: 99 permits remain unmatched ``` -There are three root causes. +There are four root causes. ## 1. The wire protocol relies on duplicated inference instead of an explicit debit @@ -98,11 +111,9 @@ reliable statement of the debit for every command. ## 2. The broker does not finalize one value and pass it end to end -The dispatcher builds batch and aggregate counts before final send eligibility is known. `Consumer.sendMessages` -then records pending-ack ownership for individual-ack subscriptions; this final admission step can reject an entry, -for example when the consumer has closed. The Shared dispatcher and command sender consume related state through -separate paths. Even when their formulas agree on the normal path, there is no finalized per-command value shared by -consumer accounting, dispatcher accounting, and serialization. +The dispatcher builds aggregate counts before final send eligibility is known. `Consumer.sendMessages` then records +pending-ack ownership; this final admission step can still reject an entry, for example when the consumer has closed. +Consumer accounting, Shared dispatcher accounting, and serialization currently have no shared post-admission result. For example, if two entries each contain ten logical messages and final admission rejects the second entry, only one command is sent. Any calculation that still uses the earlier aggregate of 20 disagrees with the actual send of ten. @@ -110,37 +121,43 @@ A wire field is useful only if it is populated from the post-admission value. ## 3. The Java client has no explicit command-to-message ownership boundary -The native batch path separately counts delivered, skipped, and failed work. It does not retain one value representing -the part of `P` still owned by command processing. Once delivery is handed to an asynchronous callback, terminal -handling and source-connection association are also spread across separate paths. +The native batch path separately counts delivered, skipped, and failed work rather than retaining the part of `P` +still owned by command processing. Once a message is handed to an asynchronous callback, terminal handling and source +association are also spread across separate paths. -For `P = 10`, if four messages enter the normal lifecycle before a later index fails, those messages should own four -units and command processing should retain six. The current delivered, skipped, and generic error counts do not +For `P = 10`, if four messages enter the normal lifecycle before a later index fails, those messages own four units +and command processing retains six. The current delivered, skipped, and generic error counts do not represent that split directly. -Adding `P` only to corrupted-message helpers would fix some early failures, but would not define who returns each unit -during partial deserialization, queue rejection, or reconnect. The client therefore needs a command-local budget as -well as the wire value. +Adding `P` only to corrupted-message helpers would fix early failures, but not partial deserialization, +asynchronous stale-epoch discard, or reconnect. The client therefore needs a command-local ownership budget as well +as the wire value. + +## 4. A failed broker write has no terminal accounting outcome + +The broker debits the finalized value before the asynchronous network write completes. On success that debit belongs +to the live consumer and is later replenished by Flow. On failure, the current write listener only logs; it does not +remove the consumer from its connection or subscription. Leaving that consumer live strands both the undelivered +permit debt and any pending-ack ownership recorded before the write. + +Re-crediting the failed value is unsafe because the transport may have accepted an unknown prefix of the write. The +unambiguous terminal outcome is to remove that broker-consumer incarnation and let normal removal/redelivery semantics +dispose of all of its connection-local accounting together. ## Current failure modes and evidence -| Failure mode | Evidence in the current path | Permit consequence | +| Failure mode | Current behavior | Permit consequence | | --- | --- | --- | -| Checksum, metadata, or decompression failure | Direct: `ConsumerImpl.messageReceived` reaches corrupted-message handling that returns one. | A command with `P > 1` loses `P - 1` permits. | -| Failure partway through native batch parsing | Direct: `ConsumerImpl.receiveIndividualMessagesFromBatch` lets already-transferred messages return individually, counts client-side skips separately, and returns one from the catch path. | The total is not tied to `P` and can be too small or too large. | -| Local queue offer rejects an asynchronously handed-off message | Direct: `ConsumerBase.enqueueMessageAndCheckBatchReceive` rolls back queue-size accounting, but has no matching terminal handling for the message and its permit. | One debited unit can be stranded. | -| Connection replacement during asynchronous handling | Race-dependent: the source `ClientCnx` is available at command receipt, but later native-batch and callback paths can read the consumer's current `cnx()`. | Credit belonging to the old broker consumer can be dropped or applied to its replacement. | -| Shared pending-ack admission rejects a batched entry after dispatch selection | Race-dependent: `Consumer.sendMessages` can null and release an entry when its `PendingAcksMap` has closed, while later consumer and dispatcher debits still originate from the earlier aggregates and the sender skips the null entry. | The server can debit permits for a command that is not sent. | - -For the partial-parsing case, suppose `D` messages were transferred and `K` deliverable messages were skipped by the -client before the failure. The current paths eventually return `D + K + 1`, regardless of how many deliverable -indexes remain. This under-returns when more than one remains and can over-return when none remains but parsing later -fails on an index that was not part of `P`. - -The first three rows follow directly from current terminal branches; the connection and closed pending-ack cases need -deterministic race-oriented tests. The table describes native-batch delivery, including common queue and connection -helpers only where that path reaches them. The asynchronous carrier lifetime in Appendix A is an implementation risk, -not a separately claimed production failure. +| Checksum, metadata, or decompression failure | `ConsumerImpl.messageReceived` reaches corrupted-message handling before `B` is available and returns one. | A command with `P > 1` loses `P - 1` permits. | +| Failure partway through native batch parsing | Transferred messages return individually, client-side skips are counted separately, and the catch path returns one. | If `D` messages were transferred and `K` were skipped, the old paths return `D + K + 1`, which is not necessarily `P`. | +| Stale epoch after asynchronous handoff | The message is released and one permit is returned through the consumer's current connection without completing the prefetch gauges. | Prefetch accounting leaks, and credit can cross into a replacement consumer. | +| Broker consumer recreation | Later paths identify the source with `ClientCnx` or read the current `cnx()`; a pooled `ClientCnx` can survive consumer replacement. | An old return can pass a connection check or race with accumulator reset. | +| Rejection after Shared dispatch selection | `Consumer.sendMessages` can remove an entry when pending-ack admission rejects it, while earlier aggregates still feed later debits. | The server can debit permits for a command that is not sent. | +| Asynchronous broker write failure | The current write listener logs the failure but leaves the consumer registered. | Undelivered debt and pending-ack ownership remain on a live consumer. | + +The checksum, parsing, stale-epoch, and write-failure rows follow directly from current terminal branches. Consumer +recreation and post-selection rejection are race-dependent and require deterministic tests. Appendix A maps these +behaviors to their current code owners. # Goals @@ -148,24 +165,30 @@ not a separately claimed production failure. - Add the actual logical-message debit `P` to `CommandMessage`. - Finalize `P` once on the broker after send eligibility is known. -- Reuse that result for consumer accounting, persistent Shared dispatcher accounting, and command serialization. +- Reuse that result for consumer accounting, both persistent Shared dispatcher implementations, and command + serialization. - Conserve `P` in the Java native-batch path across successful delivery, client-side skips, and checksum, metadata, decompression, and batch-deserialization failures. -- Bind immediate and deferred permit returns to the source client connection. +- Bind immediate and deferred permit returns to local source consumer-incarnation state, including when the same + `ClientCnx` is reused. +- Require an asynchronous message write failure to remove the affected broker consumer from its connection and + subscription. - Preserve old/new broker, Java client, and Pulsar proxy interoperability. - Validate the result end to end with persistent Shared subscriptions. ## Out of Scope -- Custom `MessagePayloadProcessor` output accounting. -- Encrypted or chunked message processing. +- Complete permit-ownership semantics for custom `MessagePayloadProcessor` output, encrypted payloads, and chunk + assembly. - Non-Java client implementations. - Permit reset, absolute synchronization, or connection-generation commands. -- General Flow/consumer-removal races and Key_Shared-specific draining behavior. +- Asynchronous broker `CommandFlow` processing racing with consumer removal, and Key_Shared-specific draining + behavior. - Changes to acknowledgment, redelivery, public receiver-queue, dispatch-rate, or message-rate semantics. - New Java public APIs, configuration, CLI commands, metrics, or broad consumer/dispatcher refactoring. -Known adjacent problems are listed in Appendix B to make the boundary explicit. +These paths are deferred rather than incompatible with the explicit debt and incarnation model; their additional +ownership rules are not defined here. Known adjacent work is listed in Appendix B. # High Level Design @@ -182,7 +205,7 @@ flowchart TB subgraph Broker A[Surviving entries] --> F[Finalize one P per command] F --> C[Debit Consumer by sum P] - F --> S[Debit persistent Shared dispatcher by sum P] + F --> S[Debit selected persistent Shared dispatcher by sum P] F --> W[Serialize P in each CommandMessage] end @@ -191,40 +214,53 @@ flowchart TB R --> Q[Command-owned budget P] Q -->|D units claimed| M[D native messages] Q -->|P minus D units remain| T[Command completion or failure] - M -->|one per terminal message| X[Return D permits to source connection] - T --> Y[Return P minus D permits to source connection] + M -->|one per terminal message| X[Return D permits to source incarnation] + T --> Y[Return P minus D permits to source incarnation] end ``` -For every successfully delivered send on a live consumer connection, the invariant is: +On every successfully written broker send that leaves the consumer live, the consumer debit is the sum of the +serialized command debits. Persistent Shared additionally uses that same sum for its dispatcher-level accounting: ```text -broker consumer debit == persistent Shared dispatcher debit == sum of all P values in the send +broker consumer debit == sum of all P values in the send +persistent Shared dispatcher debit == sum of all P values in the send +``` + +An entry rejected before handoff produces no command and no debit. A failure after handoff follows the terminal +consumer-removal rule described below instead of trying to restore this successful-send invariant by re-crediting. +For each command processed through the in-scope Java native-message path while its source consumer remains active: + +```text for each command: CommandMessage.message_permits == Java client's eventual return == P ``` -Non-batched delivery naturally uses `P = 1` and needs no separate accounting model. +Ordinary non-batched delivery naturally uses `P = 1` and needs no separate accounting model. ## Broker and Java client responsibilities The accounting contract has one authority and one owner at each stage. The broker creates the debt, the protocol -transfers its value, and the Java client conserves the debt until it is returned or its source connection disappears. +transfers its value, and the Java client conserves the debt until it is returned or its source incarnation disappears. | Component | Responsibility | | --- | --- | -| Broker admission | Decide which entries will produce commands. A rejected entry produces neither a command nor a permit debit. | -| Broker finalization | Calculate one `P` for every admitted command and its sum exactly once. All covered broker counters and serialization consume that result without recomputing it. | +| Broker finalization | Decide which entries produce commands, calculate one `P` per admitted command, and calculate their sum. A rejected entry produces neither a command nor a debit. All covered broker counters and serialization consume this result. | | Wire protocol | Carry the broker's actual per-command debit in `CommandMessage.message_permits`; this is a declaration of existing debt, not a client request. | | Java command path | Resolve and validate `P`, create `remaining = P`, transfer one unit to each accepted native message, and return every unit that was not transferred. | -| Java message lifecycle | Return exactly one transferred unit on every terminal path, including normal consumption and local rejection. | -| Connection boundary | Associate the debit and all returns with the connection that delivered the command. Connection teardown discards that connection's debt and credit; neither is moved to a replacement consumer. | -| Mixed versions | Use the explicit value when present and valid; otherwise retain the defined `ack_set`, metadata, and legacy-one fallbacks. | +| Java message lifecycle | Return exactly one unit for every in-scope transferred message, including normal consumption and stale-epoch discard. | +| Consumer-incarnation boundary | Create new local permit state for every broker-consumer incarnation, even when `ClientCnx` is reused. The state object is the incarnation token and owns only that incarnation's accumulator. Old state cannot return or accumulate credit for a replacement. | +| Broker write outcome | A successful write keeps the debit on the live consumer. A failed write must disconnect and remove that consumer; the debit and pending-ack ownership then terminate with the removed incarnation. | The complete invariant is provided only by the broker and Java responsibilities together. The protocol field alone -does not fix post-admission broker recounting or client lifecycle leaks, and either implementation PR alone preserves -mixed-version behavior without claiming the new end-to-end guarantee. +does not fix post-admission broker recounting or client lifecycle leaks. The broker and client changes are independently +compatible with old peers, but both are required to establish the new end-to-end guarantee. + +## Accounting convergence boundaries + +This design converges the arithmetic and ownership boundaries without moving payload parsing, pending acknowledgments, +queueing, or callbacks into one class. Existing code owners keep their non-accounting responsibilities. ## How the design closes the current failures @@ -234,11 +270,10 @@ broker and Java ownership rules together: | In-scope scenario | Preventing mechanism | Result with a new broker and new Java client | | --- | --- | --- | | Checksum, metadata parsing, or decompression fails before batch expansion | `P` is present in `CommandMessage` and resolved before payload processing. | The command path returns `P` without expanding the batch. | -| Native batch parsing fails after `D` messages are handed off | The command-local budget transfers only `D` units and retains `P - D`. | Messages return `D`; command failure returns `P - D`; total return is `P`. | -| A client-side skip occurs before message handoff | The skipped index never claims a unit from the command budget. | Its unit remains in `P - D` and is returned by command completion. | -| Local queue offer rejects a handed-off message | Queue rejection becomes a terminal message path that releases the message and returns its claimed unit. | The unit is neither leaked nor returned twice. | -| The source connection is replaced during asynchronous handling | The command budget and transferred messages retain the source connection. | Old credit is discarded and is never applied to the replacement consumer. | -| Pending-ack admission rejects an entry selected earlier | Broker finalizes per-entry values after admission and derives every covered debit from their sum. | The rejected entry produces no command and no permit debit. | +| Native parsing fails after `D` handoffs, or the client skips an index | The budget transfers only accepted units and retains the rest. | Messages return `D`; command completion or failure returns `P - D`; total return is `P`. | +| Stale epoch or consumer recreation, including on the same `ClientCnx` | Each message retains the source permit state; each state owns its accumulator; the event loop revalidates state before writing Flow. | Credit returns only while the source incarnation is active and can never enter the replacement accumulator. | +| Pending-ack admission rejects an entry selected earlier | Broker records a candidate `P` in the finalized result only after admission and derives every covered debit from that result's sum. | The rejected entry produces no command and no permit debit. | +| The asynchronous network write fails | The write listener disconnects the affected `Consumer`, removing it from `ServerCnx` and its subscription/dispatcher. Failure to notify the client closes the physical connection. | No live broker consumer remains with undelivered debit or stranded pending-ack ownership. | The full in-scope guarantee requires a new broker and new Java client; mixed-version behavior is defined below. @@ -248,8 +283,9 @@ The full in-scope guarantee requires a new broker and new Java client; mixed-ver ### Broker permit finalization -Logically, finalization occurs in `Consumer.sendMessages`, after pending-ack admission and before handing entries to -the asynchronous command sender. At that point the broker knows which entries will actually produce commands: +Finalization occurs in `Consumer.sendMessages`. The broker first derives a candidate `P` because pending-ack admission +needs the logical-message count. That candidate is not an accounting fact yet. Only after admission accepts the entry +does the broker record `P` in the per-send result, before handing entries to the asynchronous command sender: ```text entry is not sent => no command and no debit @@ -263,24 +299,24 @@ Finalization produces the per-entry `P` values and their sum. The same result is - the consumer's available-permit debit; - the consumer's unacked-message accounting where applicable; -- the persistent Shared dispatcher's aggregate available-permit debit; and +- the selected persistent Shared dispatcher's aggregate available-permit debit, for both + `PersistentDispatcherMultipleConsumers` and `PersistentDispatcherMultipleConsumersClassic`; and - each command's `message_permits` field. -Unacked-message accounting and permit accounting remain separate mechanisms: unacked state controls acknowledgment -backpressure, while permits control dispatch capacity. They reuse the same finalized logical-message count only so -that neither path describes messages that final admission rejected. - -Downstream layers must consume this result rather than recompute it from `B`, `ack_set`, or aggregate message counts. -Finalization is synchronous even though the network write is asynchronous: the Shared dispatcher must capture the -finalized sum as part of the send handoff, before the sender can recycle its input. The per-entry values remain valid -until the sender has serialized them. The carrier type, allocation strategy, and method signatures are implementation -details discussed in Appendix A. +Unacked state still controls acknowledgment backpressure, while permits control dispatch capacity; they only share +the finalized logical-message count. Downstream layers do not recompute it from `B`, `ack_set`, or earlier aggregates. +The per-send result owns a primitive per-entry array, its checked sum, and the asynchronous write future, so it remains +valid after the sender recycles its existing batch-size and `ack_set` inputs. -Once the send is handed to the network, the debit belongs to that source connection. An asynchronous write failure is -resolved by connection teardown, which removes the associated broker consumer and its debt; the broker does not -re-credit `P` to a replacement consumer or attempt to compensate it with a later `CommandFlow`. +Once the send is handed to the network, the debit belongs to that broker-consumer incarnation. A successful write +keeps the consumer live with that debit. An asynchronous write failure is a required terminal outcome: the write +listener calls the existing consumer-disconnect path, which removes that `Consumer` from `ServerCnx` and from its +subscription/dispatcher. The broker sends `CommandCloseConsumer` when supported; if that close notification cannot be +written, it closes the physical connection. The broker does not re-credit `P` to a replacement consumer or compensate +it with a later `CommandFlow`. -This PIP does not change dispatch-rate, byte-rate, or public message-rate statistics. +The classic dispatcher selected by `subscriptionSharedUseClassicPersistentImplementation=true` follows the same +invariant. Dispatch-rate, byte-rate, and public message-rate statistics do not change. ### Java permit resolution @@ -320,15 +356,30 @@ after `D` messages were transferred, those messages eventually return `D`, while Processing more than `P` native messages is a malformed command and follows the protocol-error behavior above. -Once a message claims one unit, ownership passes to the existing per-message lifecycle. Every terminal branch in that -lifecycle must eventually return the unit, either after normal prefetch processing or while releasing a rejected -message. The required Shared-path terminal fix includes local queue-offer rejection; it does not redesign -receiver-queue behavior or add an arbitrary permit cost to `Message`. +Once a message claims one unit, the existing per-message lifecycle owns it. Every terminal branch must return the unit +or discard it after the source broker consumer disappears. In particular, stale-epoch discard after asynchronous +handoff closes the prefetch accounting, returns through the retained source state when it is still active, and releases +the message exactly once. + +The numeric budget does not own networking state. `ConsumerImpl` creates one opaque permit-state object per +broker-consumer incarnation; the object is both the local incarnation token and that incarnation's atomic permit +accumulator. Every transferred `MessageImpl` retains that state in addition to its existing source `ClientCnx`. +Reconnect or `CommandCloseConsumer` replaces the state even when the pooled `ClientCnx` is unchanged, and never copies +the old accumulator into the replacement. + +A return can update only its retained state and succeeds only while that state is current. After a threshold drain, +the source connection's event loop revalidates exact state identity immediately before writing `CommandFlow`. A new +state is not Flow-enabled until its Subscribe succeeds. + +The connection event loop is the final ordering point for Flow and Subscribe writes: + +```text +old Flow task wins the event loop => old Flow is written before replacement Subscribe +replacement becomes visible first => old accumulator or queued Flow is discarded +``` -The budget and every message created from it retain the `ClientCnx` that delivered the command. Credit is accumulated -or sent only while that source connection is still active for the consumer. If it has been replaced, the credit is -discarded because the corresponding broker-side consumer and its debt have already been removed. Old-connection -credit is never applied to the replacement consumer. +Thus a queued old Flow either precedes the replacement Subscribe or becomes a no-op. It cannot enter the replacement +accumulator or be written after its Subscribe, and isolation does not require closing the multiplexed connection. ## Public-facing Changes @@ -345,33 +396,35 @@ message CommandMessage { optional uint64 consumer_epoch = 5; // Number of consumer permits debited by the broker for this command. - optional uint32 message_permits = 6 [default = 1]; + optional uint32 message_permits = 6; } ``` The wire rules are: - valid values are `1..Integer.MAX_VALUE`; -- the broker may omit the field when `P = 1`, and sends it when `P > 1`; +- an upgraded broker sends the field for every command, including when `P = 1`; - zero is invalid; no command is sent when nothing is deliverable; and - the value is the actual debit, not necessarily the original batch size. No protocol-version bump or feature flag is required. Old protobuf readers ignore the optional field, so an upgraded -broker can send it without detecting client support. New Java clients use field presence and the fallback rules. +broker can send it without detecting client support. New Java clients use field presence to distinguish an explicit +debit from an old-broker command that requires the compatibility fallback rules. # Monitoring -No metric is added. Existing consumer `availablePermits`, receive-failure statistics, and dispatch progress can help -diagnose the behavior. Tests verify the accounting invariant directly and verify that repeated corrupted batches do -not stall a Shared consumer. +No metric is added. Existing `pulsar_consumer_available_permits` or +`pulsar.broker.consumer.permit.count`, receive-failure statistics, and dispatch progress can help diagnose a stall, +but none independently proves that both sides agreed on `P`. Tests therefore verify the accounting invariant directly +and verify that repeated corrupted batches do not stall a Shared consumer. # Security Considerations The broker-to-client field grants no client authority and does not change authentication, authorization, or tenant isolation. -The Java client must reject zero and unsigned values above `Integer.MAX_VALUE` and must not return more than `P` for -one command. Permit accumulation uses checked arithmetic; overflow is handled as a protocol error on the source -connection rather than wrapping into negative or unrelated Flow credit. +The Java client must reject zero and unsigned values above `Integer.MAX_VALUE`. The in-scope native path must not +return more than `P` for one command. Permit accumulation uses checked arithmetic; overflow is handled as a protocol +error on the source connection rather than wrapping into negative or unrelated Flow credit. # Backward & Forward Compatibility @@ -382,9 +435,10 @@ connection rather than wrapping into negative or unrelated Flow credit. | Old | New | The client falls back to `ack_set`, parsed batch metadata, then legacy one. | | Old | Old | No change. | -Omitting `P = 1` is unambiguous. A complete one-message command falls back to one, while a partial batch with one -deliverable index carries an `ack_set` with cardinality one. An old client already returns one permit per successfully -delivered native message, so an upgraded broker does not change its normal path. +An upgraded broker does not special-case `P = 1`: it explicitly declares the debit for both a complete one-message +command and a partial batch with one deliverable index. Field absence is reserved for compatibility with old brokers +or intermediaries that remove unknown fields. An old client ignores the new field and already returns one permit per +successfully delivered native message, so an upgraded broker does not change its normal path. Pulsar proxy paths forward the original message frame without re-serializing `CommandMessage`. A third-party proxy that drops unknown fields remains wire-compatible through the fallback, but cannot preserve the new early-failure @@ -416,54 +470,61 @@ rollback. - **Add a reset or absolute-permit command:** requires ordering, idempotency, in-flight command, and connection- generation rules. Attaching the debit to the command that creates it is smaller. - **Assign arbitrary permit costs to `Message`:** unnecessary for native batches, where each accepted logical message - owns exactly one unit. Custom processors need a separate entry-level completion design. -- **Separate broker and client PIPs:** risks defining different debit and return semantics. Implementation can still - be staged under one end-to-end protocol contract. + owns exactly one unit. Deferred processing paths require their own ownership rules. +- **Separate broker and client PIPs:** risks defining different debit and return semantics. The selected design keeps + two implementation layers under one end-to-end protocol contract and one pull request. # General Notes -After the PIP is accepted, implementation is split into two focused pull requests: +The implementation is organized as one end-to-end pull request with two commit layers: 1. **Protocol and broker:** add the field, finalize one per-entry value, align consumer and persistent Shared - aggregate debits, serialize that value, and test old-reader compatibility. -2. **Java native batch:** implement resolution and fallbacks, command-local ownership, source-connection binding, the - narrow terminal fixes, and the persistent Shared end-to-end test. + aggregate debits in both modern and classic implementations, serialize that value, make failed writes remove the + affected consumer, and test old-reader compatibility. +2. **Java native batch:** implement resolution and fallbacks, command-local ownership, consumer-incarnation binding, + the narrow terminal fixes, and the persistent Shared end-to-end test. -The first PR does not introduce client lifecycle abstractions. The second PR does not redesign dispatcher selection -or Flow/removal ordering. Neither PR alone establishes the end-to-end guarantee. +Keeping both layers in one PR makes the complete invariant reviewable together; commits retain the same boundary for +review and later backporting. The required test matrix covers: -- full and partial native batches, including `P = 1` omission; +- full and partial native batches, including explicit `P = 1` and old-broker field absence; - filtering or pending-ack rejection before broker finalization; -- consumer and persistent Shared debits equal the sum of the per-command values, while each command carries its `P`; -- asynchronous ownership and recycling of the broker's per-entry carrier; +- consumer and persistent Shared debits equal the sum of the per-command values in both modern and classic + dispatchers, while each command carries its `P`; +- asynchronous lifetime of finalized per-entry values while existing sender inputs are recycled; +- asynchronous write failure removes the affected consumer from its connection and subscription/dispatcher; - explicit-field validation and old-broker `ack_set`/metadata fallbacks; - checksum, metadata, decompression, and mid-batch deserialization failures; -- native skips, queue rejection, and source-connection replacement; and +- native skips, stale-epoch discard, different-connection replacement, same-`ClientCnx` recreation, a concurrent + old-return/recreation race, and a queued old Flow that executes after same-connection state replacement; and - repeated corrupted batches on a persistent Shared subscription without dispatch stall. -# Appendix A: Current implementation map +# Appendix A: Implementation map and convergence points -This appendix records where the current behavior is distributed. It is supporting evidence, not an additional -implementation scope. +This appendix records where the current behavior is distributed and identifies the two narrow accounting components +introduced by this design. It is supporting detail, not additional implementation scope. -| Component | Current responsibility | Relevance to this PIP | +| Component | Responsibility | Relevance to this PIP | | --- | --- | --- | | `MessageMetadata.num_messages_in_batch` | Stores original payload cardinality `B`. | Payload structure is not always the actual debit and is unavailable before metadata parsing. | | `AbstractBaseDispatcher.filterEntriesForConsumer` | Extracts `B`, `ack_set`, and aggregate counts. | Runs before final pending-ack admission. | | `Consumer.sendMessages` | Performs pending-ack admission and debits consumer/unacked counts. | Must finalize the values for entries that survive admission. | -| `PersistentDispatcherMultipleConsumers` | Debits aggregate `totalAvailablePermits`. | Must consume the same finalized aggregate rather than recalculate it. | +| PIP-491 `SendMessagesResult` | Carries one primitive per-entry `P` array, its checked sum, and the asynchronous write result for one send. | Establishes the broker finalization boundary consumed by accounting and serialization. | +| `PersistentDispatcherMultipleConsumers` | Default persistent Shared implementation; debits aggregate `totalAvailablePermits`. | Must consume the same finalized aggregate rather than recalculate it. | +| `PersistentDispatcherMultipleConsumersClassic` | Configurable classic persistent Shared implementation; independently debits aggregate `totalAvailablePermits`. | Must consume the same finalized aggregate and preserve the invariant when the rollback setting selects it. | | `PulsarCommandSenderImpl.sendMessagesToConsumer` | Serializes surviving entries and `ack_set`. | Must serialize the finalized per-entry value without recomputation. | | `EntryBatchIndexesAcks` | Carries mutable pooled per-entry `ack_set` state. | The sender owns and recycles it asynchronously, so finalized values need an explicit lifetime and must not be recomputed after handoff. | | `ConsumerImpl.messageReceived` | Verifies checksum, parses metadata, decompresses, and chooses the payload path. | Must resolve `P` before payload-dependent failures. | | `ConsumerImpl.receiveIndividualMessagesFromBatch` | Splits native batches and counts skipped messages. | Must transfer units from the command budget instead of reconstructing returns. | -| `ConsumerImpl.executeNotifyCallback` | Performs asynchronous callback handling. | A transferred unit needs an exact terminal outcome on its source connection. | -| `ConsumerBase.enqueueMessageAndCheckBatchReceive` | Admits messages to the local queue. | Queue-offer rejection must release the native message and its unit. | +| `ConsumerImpl.executeNotifyCallback` | Performs asynchronous callback handling. | A transferred unit needs an exact terminal outcome on its source permit state. | +| PIP-491 `MessagePermitAccounting` | Resolves and validates `P` and owns the numeric command budget. | Centralizes permit inference and conservation without coupling the budget to connection or queue state. | +| PIP-491 local `ConsumerPermitState` | Acts as the identity token and permit accumulator for one broker-consumer incarnation. | Keeps incarnation identity separate from the existing message-source `ClientCnx` and prevents old asynchronous returns from crossing a same-connection consumer recreation. | -The broker carrier introduced by this PIP should continue the existing low-allocation approach: use primitive pooled -storage where practical and avoid allocating one object per dispatched entry. This performance constraint does not -change the protocol contract. +The broker result introduced by this PIP uses one primitive array per send and avoids allocating one object per +dispatched entry. It does not need to be pooled; existing pooled sender inputs retain their current recycling rules. +This low-allocation shape does not change the protocol contract. # Appendix B: Related work outside this PIP @@ -475,10 +536,10 @@ The following known areas have different ordering or ownership requirements and - encrypted payload and chunk-assembly lifecycles; - non-Java client prefetch implementations; - multi-topic parent and zero-queue specializations beyond behavior reached through the native child consumer; -- channel write failure and teardown for commands that were not successfully delivered; and - broader entry, batch, and logical-message unit consistency in statistics and rate limiting. -These areas may reuse the explicit field later, but they do not block the native Java Shared invariant. +These areas remain compatible with the explicit debt and incarnation model but are outside the native Java Shared +invariant defined by this PIP. # Links @@ -490,9 +551,11 @@ These areas may reuse the explicit field later, but they do not block the native [AbstractBaseDispatcher](../pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java), [Consumer](../pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java), [persistent Shared dispatcher](../pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java), + [classic persistent Shared dispatcher](../pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassic.java), and [command sender](../pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSenderImpl.java) * Current Java paths: [ConsumerImpl](../pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java) and [ConsumerBase](../pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java) +* Accounting convergence types: broker `SendMessagesResult` and Java `MessagePermitAccounting`. * Mailing List discussion thread: * Mailing List voting thread: From cc231bf1f80b09fb5a901bbd578774e63c6d5bb1 Mon Sep 17 00:00:00 2001 From: void-ptr974 Date: Sun, 16 Aug 2026 13:49:50 +0800 Subject: [PATCH 4/7] [improve][pip] Clarify PIP-491 title and purpose --- pip/pip-491.md | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/pip/pip-491.md b/pip/pip-491.md index 06ad80629f58c..82311341c0804 100644 --- a/pip/pip-491.md +++ b/pip/pip-491.md @@ -1,4 +1,4 @@ -# PIP-491: Explicit permit accounting for batched message delivery +# PIP-491: Prevent Delivery Stalls by Making the Client Return Exactly the Permits Used by the Broker # Background knowledge @@ -51,9 +51,10 @@ to the removed broker consumer must not be compensated to its replacement, even # Motivation -For each sent `CommandMessage`, the broker must debit `P`, but the command does not carry `P`. The Java client must -infer it from `ack_set` and the payload. If the payload cannot be parsed, the client may not know `P`; if only part of -a batch enters the message lifecycle, the current code does not preserve how ownership of `P` was split. +For each sent `CommandMessage`, the broker uses `P` consumer permits and the Java client must eventually return +exactly `P`. The command does not carry `P`, so the client has to infer it from `ack_set` and the payload. If the +payload cannot be parsed or only part of a batch enters the message lifecycle, the client can return a different +count from the one used by the broker. The current protocol therefore relies on duplicated computation across broker layers and the client. The dispatcher, consumer, and sender derive related counts at different points in the send lifecycle, while the client independently From 558999e4d3278fcec9a4ee7d7cf20a9415d284ed Mon Sep 17 00:00:00 2001 From: void-ptr974 Date: Sun, 16 Aug 2026 17:42:47 +0800 Subject: [PATCH 5/7] [improve][pip] Clarify negative Shared permit balances Assisted-by: OpenAI Codex --- pip/pip-491.md | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/pip/pip-491.md b/pip/pip-491.md index 82311341c0804..fac3c80a9bb2c 100644 --- a/pip/pip-491.md +++ b/pip/pip-491.md @@ -41,6 +41,13 @@ For example: The partial example shows why `B` cannot be reused as the debit: the payload still contains eight original indexes, but only three logical messages are eligible for this delivery. +The broker's available-permit balance is distinct from `P` and may temporarily be negative. Dispatch operates on +indivisible entries, so a positive balance of one can admit a batch with `P = 10`; after the send, both the consumer +balance and, when it has no other consumers, the persistent Shared dispatcher balance are `-9`. This is outstanding +batch debt, not an invalid permit value. The consumer is not eligible for another send until returned credit makes its +balance positive again. Implementations must debit and later credit the full `P`; clamping or resetting the balance to +zero would discard the debt and grant excess capacity when the client returns those ten permits. + The acknowledgment path is separate from permit accounting. Acknowledgment controls cursor progress and redelivery; permits control subsequent dispatch. Acknowledging an entry does not repair a missing permit, and returning a permit does not acknowledge a message. @@ -494,6 +501,8 @@ The required test matrix covers: - filtering or pending-ack rejection before broker finalization; - consumer and persistent Shared debits equal the sum of the per-command values in both modern and classic dispatchers, while each command carries its `P`; +- batch overshoot from one available permit to a negative broker balance, followed by exact recovery when the client + returns the full `P`, in both modern and classic persistent Shared dispatchers; - asynchronous lifetime of finalized per-entry values while existing sender inputs are recycled; - asynchronous write failure removes the affected consumer from its connection and subscription/dispatcher; - explicit-field validation and old-broker `ack_set`/metadata fallbacks; From 7f5aec345b93b2895b442c04ce2cc27e139253c3 Mon Sep 17 00:00:00 2001 From: void-ptr974 Date: Sun, 16 Aug 2026 22:30:35 +0800 Subject: [PATCH 6/7] [improve][pip] Define the client Flow trust boundary Assisted-by: OpenAI Codex --- pip/pip-491.md | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/pip/pip-491.md b/pip/pip-491.md index fac3c80a9bb2c..7d33dc56c9393 100644 --- a/pip/pip-491.md +++ b/pip/pip-491.md @@ -189,6 +189,7 @@ behaviors to their current code owners. - Complete permit-ownership semantics for custom `MessagePayloadProcessor` output, encrypted payloads, and chunk assembly. - Non-Java client implementations. +- Broker-side attestation or reconciliation of arbitrary client `CommandFlow` values. - Permit reset, absolute synchronization, or connection-generation commands. - Asynchronous broker `CommandFlow` processing racing with consumer removal, and Key_Shared-specific draining behavior. @@ -261,6 +262,18 @@ transfers its value, and the Java client conserves the debt until it is returned | Consumer-incarnation boundary | Create new local permit state for every broker-consumer incarnation, even when `ClientCnx` is reused. The state object is the incarnation token and owns only that incarnation's accumulator. Old state cannot return or accumulate credit for a replacement. | | Broker write outcome | A successful write keeps the debit on the live consumer. A failed write must disconnect and remove that consumer; the debit and pending-ack ownership then terminate with the removed incarnation. | +This is a correctness guarantee for the new protocol-compliant Java implementation, not broker attestation of client +behavior. `CommandFlow` currently carries only a `consumer_id` and an incremental `messagePermits` value. It has no +delivery identifier, incarnation, or cumulative sequence that lets the broker prove that a Flow increment corresponds +to previously dispatched debt. A client that returns too little eventually stalls itself; a buggy or malicious client +can return too much and weaken its own receiver-side backpressure. This trust boundary already exists and is not +expanded by declaring `P`. + +Broker-side detection would require a separate protocol design, such as incarnation-scoped cumulative returns or +broker-issued delivery-credit identifiers, while also distinguishing initial/window-growth grants from returned +credit. Such a mechanism could reject stale, duplicate, or excessive Flow increments, but still could not prove that +the application actually processed a message. It is outside this PIP's Java correctness scope. + The complete invariant is provided only by the broker and Java responsibilities together. The protocol field alone does not fix post-admission broker recounting or client lifecycle leaks. The broker and client changes are independently compatible with old peers, but both are required to establish the new end-to-end guarantee. @@ -429,7 +442,8 @@ and verify that repeated corrupted batches do not stall a Shared consumer. # Security Considerations The broker-to-client field grants no client authority and does not change authentication, authorization, or tenant -isolation. +isolation. The broker continues to trust a connected client to send protocol-compliant Flow increments; this PIP makes +the new Java client's accounting deterministic and testable but does not make `CommandFlow` self-authenticating. The Java client must reject zero and unsigned values above `Integer.MAX_VALUE`. The in-scope native path must not return more than `P` for one command. Permit accumulation uses checked arithmetic; overflow is handled as a protocol error on the source connection rather than wrapping into negative or unrelated Flow credit. @@ -509,6 +523,8 @@ The required test matrix covers: - checksum, metadata, decompression, and mid-batch deserialization failures; - native skips, stale-epoch discard, different-connection replacement, same-`ClientCnx` recreation, a concurrent old-return/recreation race, and a queued old Flow that executes after same-connection state replacement; and +- deterministic randomized native batches whose decoded `CommandFlow.messagePermits` sum equals the explicit `P` + sum after all delivered messages and immediate skips reach terminal handling; and - repeated corrupted batches on a persistent Shared subscription without dispatch stall. # Appendix A: Implementation map and convergence points From 70d8df458ff164c1d7fed3e62dcefb18ded13bb6 Mon Sep 17 00:00:00 2001 From: void-ptr974 Date: Tue, 18 Aug 2026 13:28:40 +0800 Subject: [PATCH 7/7] [improve][pip] Address PIP-491 review feedback Assisted-by: OpenAI Codex --- pip/pip-491.md | 329 ++++++++++++++++++++++++++++++++----------------- 1 file changed, 214 insertions(+), 115 deletions(-) diff --git a/pip/pip-491.md b/pip/pip-491.md index 7d33dc56c9393..4cc39e216ba0e 100644 --- a/pip/pip-491.md +++ b/pip/pip-491.md @@ -7,11 +7,10 @@ capacity, the broker consumes that capacity while dispatching, and the client re prefetch path. In the consumer flow accounting covered by this PIP, one permit represents one logical message, not one BookKeeper entry. -The broker represents this capacity at two levels for persistent Shared subscriptions. Each `Consumer` tracks its -own available permits, while the Shared dispatcher tracks aggregate `totalAvailablePermits` across its consumers to -decide whether dispatch can continue. A send must therefore debit both levels by the same number of logical messages. -If they use different counts, the dispatcher and the selected consumer disagree about remaining capacity even when -the network write itself succeeds. +Each broker `Consumer` tracks its own available permits. Dispatchers that schedule across multiple consumers can also +track aggregate available permits to decide whether dispatch can continue. A send must debit the selected consumer +and every applicable dispatcher aggregate by the same number of logical messages. If they use different counts, the +dispatcher and selected consumer disagree about remaining capacity even when the network write itself succeeds. A native Pulsar batch stores multiple logical messages in one entry. The payload metadata field `num_messages_in_batch` describes the original batch cardinality so the broker can account for the entry and the @@ -34,8 +33,8 @@ For example: | Entry | `B` | Deliverable `ack_set` indexes | `P` | | --- | ---: | --- | ---: | -| Non-batched message | 1 | absent | 1 | -| Complete native batch | 8 | absent | 8 | +| Non-batched message | 1 | empty | 1 | +| Complete native batch | 8 | empty | 8 | | Partial native batch | 8 | `{0, 2, 5}` | 3 | The partial example shows why `B` cannot be reused as the debit: the payload still contains eight original indexes, @@ -58,10 +57,10 @@ to the removed broker consumer must not be compensated to its replacement, even # Motivation -For each sent `CommandMessage`, the broker uses `P` consumer permits and the Java client must eventually return -exactly `P`. The command does not carry `P`, so the client has to infer it from `ack_set` and the payload. If the -payload cannot be parsed or only part of a batch enters the message lifecycle, the client can return a different -count from the one used by the broker. +For each sent `CommandMessage`, the broker uses `P` consumer permits. While the source consumer incarnation remains +active, the in-scope Java path must eventually return exactly `P`. The command does not carry `P`, so the client has +to infer it from `ack_set` and the payload. If the payload cannot be parsed or only part of a batch enters the message +lifecycle, the client can return a different count from the one used by the broker. The current protocol therefore relies on duplicated computation across broker layers and the client. The dispatcher, consumer, and sender derive related counts at different points in the send lifecycle, while the client independently @@ -107,14 +106,14 @@ sequenceDiagram Note over B,J: 99 permits remain unmatched ``` -There are four root causes. +There are three root causes. ## 1. The wire protocol relies on duplicated inference instead of an explicit debit `num_messages_in_batch` cannot be the permit contract because it is inside the payload metadata. It is unavailable when checksum or metadata parsing fails, and it remains `B` when a partial redelivery has `P < B`. -`ack_set` is sufficient for a partial batch but is normally absent for a complete batch. Neither existing field is a +`ack_set` is sufficient for a partial batch but is normally empty for a complete batch. Neither existing field is a reliable statement of the debit for every command. ## 2. The broker does not finalize one value and pass it end to end @@ -141,31 +140,21 @@ Adding `P` only to corrupted-message helpers would fix early failures, but not p asynchronous stale-epoch discard, or reconnect. The client therefore needs a command-local ownership budget as well as the wire value. -## 4. A failed broker write has no terminal accounting outcome - -The broker debits the finalized value before the asynchronous network write completes. On success that debit belongs -to the live consumer and is later replenished by Flow. On failure, the current write listener only logs; it does not -remove the consumer from its connection or subscription. Leaving that consumer live strands both the undelivered -permit debt and any pending-ack ownership recorded before the write. - -Re-crediting the failed value is unsafe because the transport may have accepted an unknown prefix of the write. The -unambiguous terminal outcome is to remove that broker-consumer incarnation and let normal removal/redelivery semantics -dispose of all of its connection-local accounting together. - ## Current failure modes and evidence | Failure mode | Current behavior | Permit consequence | | --- | --- | --- | -| Checksum, metadata, or decompression failure | `ConsumerImpl.messageReceived` reaches corrupted-message handling before `B` is available and returns one. | A command with `P > 1` loses `P - 1` permits. | +| Checksum or metadata failure | `ConsumerImpl.messageReceived` reaches corrupted-message handling before `B` is available and returns one. | A command with `P > 1` loses `P - 1` permits. | +| Decompression failure | `B` is available, but the generic corrupted-message path still returns one. | A command with `P > 1` loses `P - 1` permits even though the client has enough information. | | Failure partway through native batch parsing | Transferred messages return individually, client-side skips are counted separately, and the catch path returns one. | If `D` messages were transferred and `K` were skipped, the old paths return `D + K + 1`, which is not necessarily `P`. | +| Client-side skip or parent handoff | Duplicate, start-position, compaction, dead-letter, child-consumer, and parent-consumer branches can each return directly. | A permit can be lost or returned twice unless one owner handles each terminal outcome. | | Stale epoch after asynchronous handoff | The message is released and one permit is returned through the consumer's current connection without completing the prefetch gauges. | Prefetch accounting leaks, and credit can cross into a replacement consumer. | | Broker consumer recreation | Later paths identify the source with `ClientCnx` or read the current `cnx()`; a pooled `ClientCnx` can survive consumer replacement. | An old return can pass a connection check or race with accumulator reset. | | Rejection after Shared dispatch selection | `Consumer.sendMessages` can remove an entry when pending-ack admission rejects it, while earlier aggregates still feed later debits. | The server can debit permits for a command that is not sent. | -| Asynchronous broker write failure | The current write listener logs the failure but leaves the consumer registered. | Undelivered debt and pending-ack ownership remain on a live consumer. | -The checksum, parsing, stale-epoch, and write-failure rows follow directly from current terminal branches. Consumer -recreation and post-selection rejection are race-dependent and require deterministic tests. Appendix A maps these -behaviors to their current code owners. +The checksum, decompression, parsing, skip, and stale-epoch rows follow directly from current terminal branches. +Consumer recreation and post-selection rejection are race-dependent and require deterministic tests. Appendix A +maps these behaviors to their current code owners. # Goals @@ -173,31 +162,33 @@ behaviors to their current code owners. - Add the actual logical-message debit `P` to `CommandMessage`. - Finalize `P` once on the broker after send eligibility is known. -- Reuse that result for consumer accounting, both persistent Shared dispatcher implementations, and command - serialization. +- Reuse that result for consumer accounting, command serialization, and every dispatcher aggregate debited for the + send. - Conserve `P` in the Java native-batch path across successful delivery, client-side skips, and checksum, metadata, decompression, and batch-deserialization failures. +- Define one permit-return owner for direct returns, queue removal, listener and parent handoff, and zero-queue + terminal paths reached by the built-in Java client. - Bind immediate and deferred permit returns to local source consumer-incarnation state, including when the same `ClientCnx` is reused. -- Require an asynchronous message write failure to remove the affected broker consumer from its connection and - subscription. - Preserve old/new broker, Java client, and Pulsar proxy interoperability. - Validate the result end to end with persistent Shared subscriptions. ## Out of Scope -- Complete permit-ownership semantics for custom `MessagePayloadProcessor` output, encrypted payloads, and chunk - assembly. -- Non-Java client implementations. +- Exact output-cardinality semantics after control transfers to a custom `MessagePayloadProcessor`, chunk assembly, + or a crypto failure action. Successful decryption that returns to the built-in native path remains in scope. +- Non-Java client implementations, which require client-specific follow-up work. - Broker-side attestation or reconciliation of arbitrary client `CommandFlow` values. - Permit reset, absolute synchronization, or connection-generation commands. -- Asynchronous broker `CommandFlow` processing racing with consumer removal, and Key_Shared-specific draining - behavior. -- Changes to acknowledgment, redelivery, public receiver-queue, dispatch-rate, or message-rate semantics. +- Asynchronous broker `CommandFlow` processing racing with consumer removal. +- New broker network-write failure handling; existing transport cleanup remains unchanged. +- Changes to Key_Shared routing or draining policy. Accounting for the resulting send remains in scope. +- Changes to acknowledgment, redelivery, public receiver-queue, dispatch-rate, or message-rate policy. Permit outcomes + of existing client removal and redelivery paths remain in scope. - New Java public APIs, configuration, CLI commands, metrics, or broad consumer/dispatcher refactoring. -These paths are deferred rather than incompatible with the explicit debt and incarnation model; their additional -ownership rules are not defined here. Known adjacent work is listed in Appendix B. +These exclusions do not change wire compatibility. Specialized payload paths retain their current cardinality and +ownership rules as detailed below. Known adjacent work is listed in Appendix B. # High Level Design @@ -214,7 +205,7 @@ flowchart TB subgraph Broker A[Surviving entries] --> F[Finalize one P per command] F --> C[Debit Consumer by sum P] - F --> S[Debit selected persistent Shared dispatcher by sum P] + F --> S[Debit applicable dispatcher aggregate by sum P] F --> W[Serialize P in each CommandMessage] end @@ -229,15 +220,14 @@ flowchart TB ``` On every successfully written broker send that leaves the consumer live, the consumer debit is the sum of the -serialized command debits. Persistent Shared additionally uses that same sum for its dispatcher-level accounting: +serialized command debits. Any dispatcher aggregate debited for that send uses the same sum: ```text broker consumer debit == sum of all P values in the send -persistent Shared dispatcher debit == sum of all P values in the send +applicable dispatcher aggregate debit == sum of all P values in the send ``` -An entry rejected before handoff produces no command and no debit. A failure after handoff follows the terminal -consumer-removal rule described below instead of trying to restore this successful-send invariant by re-crediting. +An entry rejected before handoff produces no command and no debit. For each command processed through the in-scope Java native-message path while its source consumer remains active: @@ -260,7 +250,6 @@ transfers its value, and the Java client conserves the debt until it is returned | Java command path | Resolve and validate `P`, create `remaining = P`, transfer one unit to each accepted native message, and return every unit that was not transferred. | | Java message lifecycle | Return exactly one unit for every in-scope transferred message, including normal consumption and stale-epoch discard. | | Consumer-incarnation boundary | Create new local permit state for every broker-consumer incarnation, even when `ClientCnx` is reused. The state object is the incarnation token and owns only that incarnation's accumulator. Old state cannot return or accumulate credit for a replacement. | -| Broker write outcome | A successful write keeps the debit on the live consumer. A failed write must disconnect and remove that consumer; the debit and pending-ack ownership then terminate with the removed incarnation. | This is a correctness guarantee for the new protocol-compliant Java implementation, not broker attestation of client behavior. `CommandFlow` currently carries only a `consumer_id` and an incremental `messagePermits` value. It has no @@ -281,7 +270,9 @@ compatible with old peers, but both are required to establish the new end-to-end ## Accounting convergence boundaries This design converges the arithmetic and ownership boundaries without moving payload parsing, pending acknowledgments, -queueing, or callbacks into one class. Existing code owners keep their non-accounting responsibilities. +queueing, or callbacks into one class. Existing code owners keep their non-accounting responsibilities, but must +consume finalized `P`, the resolved command budget, and retained source-incarnation state instead of independently +reconstructing permit debt. ## How the design closes the current failures @@ -290,11 +281,12 @@ broker and Java ownership rules together: | In-scope scenario | Preventing mechanism | Result with a new broker and new Java client | | --- | --- | --- | -| Checksum, metadata parsing, or decompression fails before batch expansion | `P` is present in `CommandMessage` and resolved before payload processing. | The command path returns `P` without expanding the batch. | +| Checksum or metadata parsing fails before `B` is available | `P` is present in `CommandMessage` and resolved before payload processing. | The command path returns `P` without depending on payload metadata. | +| Decompression fails after `B` is available | The resolved command budget remains authoritative after metadata parsing. | The command path returns `P`, rather than the generic one-permit fallback. | | Native parsing fails after `D` handoffs, or the client skips an index | The budget transfers only accepted units and retains the rest. | Messages return `D`; command completion or failure returns `P - D`; total return is `P`. | +| A direct skip or parent handoff reaches more than one return site | Each unit has one command or message owner, and a terminal action consumes that ownership. | An outer cleanup or wrapper cannot return the same unit again. | | Stale epoch or consumer recreation, including on the same `ClientCnx` | Each message retains the source permit state; each state owns its accumulator; the event loop revalidates state before writing Flow. | Credit returns only while the source incarnation is active and can never enter the replacement accumulator. | | Pending-ack admission rejects an entry selected earlier | Broker records a candidate `P` in the finalized result only after admission and derives every covered debit from that result's sum. | The rejected entry produces no command and no permit debit. | -| The asynchronous network write fails | The write listener disconnects the affected `Consumer`, removing it from `ServerCnx` and its subscription/dispatcher. Failure to notify the client closes the physical connection. | No live broker consumer remains with undelivered debit or stranded pending-ack ownership. | The full in-scope guarantee requires a new broker and new Java client; mixed-version behavior is defined below. @@ -310,18 +302,18 @@ does the broker record `P` in the per-send result, before handing entries to the ```text entry is not sent => no command and no debit -ack_set is absent => P = B -ack_set is present => P = deliverable set-bit count within B +ack_set is empty => P = B +ack_set is non-empty => P = deliverable set-bit count within B ``` -An entry with `P = 0` is not sent. A serialized `CommandMessage` always represents at least one logical message. +An entry with `P = 0` is not sent, debited, or recorded as pending-ack ownership. A serialized `CommandMessage` +always represents at least one logical message. Finalization produces the per-entry `P` values and their sum. The same result is used for: - the consumer's available-permit debit; - the consumer's unacked-message accounting where applicable; -- the selected persistent Shared dispatcher's aggregate available-permit debit, for both - `PersistentDispatcherMultipleConsumers` and `PersistentDispatcherMultipleConsumersClassic`; and +- every applicable dispatcher aggregate available-permit debit; and - each command's `message_permits` field. Unacked state still controls acknowledgment backpressure, while permits control dispatch capacity; they only share @@ -329,15 +321,43 @@ the finalized logical-message count. Downstream layers do not recompute it from The per-send result owns a primitive per-entry array, its checked sum, and the asynchronous write future, so it remains valid after the sender recycles its existing batch-size and `ack_set` inputs. -Once the send is handed to the network, the debit belongs to that broker-consumer incarnation. A successful write -keeps the consumer live with that debit. An asynchronous write failure is a required terminal outcome: the write -listener calls the existing consumer-disconnect path, which removes that `Consumer` from `ServerCnx` and from its -subscription/dispatcher. The broker sends `CommandCloseConsumer` when supported; if that close notification cannot be -written, it closes the physical connection. The broker does not re-credit `P` to a replacement consumer or compensate -it with a later `CommandFlow`. +All `Consumer` send entry points, including overloads carrying consumer epochs or sticky-key hashes, expose the same +finalized values and sum to callers that own aggregate accounting. + +Once a command is successfully written, the debit belongs to that broker-consumer incarnation. This PIP does not add +new handling for asynchronous network-write failure or speculative re-crediting after an ambiguous write; existing +transport cleanup remains unchanged. + +#### Dispatcher aggregate coverage + +Every dispatcher that subtracts a logical-message count from aggregate available permits uses the finalized sum for +the commands actually emitted: + +| Dispatcher | Required aggregate debit | +| --- | --- | +| `PersistentDispatcherMultipleConsumers` | Finalized sum. | +| `PersistentDispatcherMultipleConsumersClassic` | Finalized sum. | +| `PersistentStickyKeyDispatcherMultipleConsumers` | Finalized sum, including its overridden sticky-key send loop. | +| `PersistentStickyKeyDispatcherMultipleConsumersClassic` | Finalized sum. | +| `NonPersistentDispatcherMultipleConsumers` | Finalized sum. | +| `NonPersistentStickyKeyDispatcherMultipleConsumers` | Finalized sum. | + +The rule applies to normal, replay, and chunk-specific loops that debit an aggregate. Persistent and non-persistent +single-active dispatchers have no such multi-consumer aggregate; they still debit the selected consumer and serialize +the finalized per-command value. This changes no Key_Shared routing or draining decision. + +#### `avgMessagesPerEntry` and precise dispatcher flow control + +The permit unit remains one logical message whether `preciseDispatcherFlowControl` is enabled or disabled. After +finalization, `avgMessagesPerEntry` is updated as: + +```text +sum(finalized P values) / number of emitted entries +``` -The classic dispatcher selected by `subscriptionSharedUseClassicPersistentImplementation=true` follows the same -invariant. Dispatch-rate, byte-rate, and public message-rate statistics do not change. +When no entry is emitted, the average is not updated. This statistic therefore describes final deliverable logical +messages per emitted entry, and the existing `ConsumerStats.avgMessagesPerEntry` field exposes that corrected value. +Existing dispatch-rate, byte-rate, and message-rate semantics otherwise remain unchanged. ### Java permit resolution @@ -348,14 +368,20 @@ For the native path, the Java client resolves `P` in this order: 3. Otherwise, after parsing metadata, use `num_messages_in_batch`. 4. If an old-broker command fails before either source is available, use the legacy value of one. -A present `message_permits` value is authoritative for flow accounting once validated. It must be in the range -`1..Integer.MAX_VALUE`; Java must interpret and validate the protobuf `uint32` value using unsigned conversion. +A present `message_permits` value is authoritative for flow accounting once validated. Every resolved value must be +positive. Java interprets the protobuf `uint32` using unsigned conversion and accepts an explicit value only in +`1..Integer.MAX_VALUE`. -After native metadata is available, an explicit value must equal `B` for a complete batch or the deliverable -`ack_set` cardinality for a partial batch. This validation happens before any native message is transferred to the -per-message lifecycle. An invalid explicit value is a protocol error: the client sends no Flow credit and closes the +After native metadata is available, `B` must be positive. A complete batch requires `P = B`; a partial batch requires +`P` to equal the deliverable `ack_set` cardinality, with every set bit in `[0, B)`. A non-empty all-zero `ack_set`, an +out-of-range set bit, or native expansion beyond `P` is malformed. Validation occurs before accepting a native +message whenever the required metadata is available. A malformed command returns no Flow credit and closes the source connection so the associated broker-side debt is discarded. +Before metadata is available, the client can validate only the explicit field's unsigned range or count an +old-broker `ack_set`; full consistency validation occurs after metadata parsing. An explicit range-valid `P` remains +available to checksum and metadata-failure paths. An absent-field early fallback is necessarily best effort. + The final fallback cannot recover a complete-batch debit from an old broker when failure occurs before metadata parsing. Exact conservation for that case requires an upgraded broker. The client retains the legacy one-permit behavior rather than closing a multiplexed connection for this old-protocol ambiguity. @@ -365,22 +391,51 @@ behavior rather than closing a multiplexed connection for this old-protocol ambi Native command processing starts with `remaining = P`. Indexes excluded by the broker's `ack_set` are not part of `P` and never interact with the budget. For each -deliverable index, immediately before its logical message is accepted into the existing per-message lifecycle, the -client claims one unit by decrementing `remaining`. If the transfer itself fails, the claim is restored. +deliverable index, one unit transfers from the command budget when handoff to the existing per-message lifecycle +succeeds. Constructing a message object alone is not acceptance. If an implementation reserves a unit before calling +the handoff, failure restores that reservation. -A deliverable index that the client later skips as duplicate, compacted-out, or otherwise ineligible does not claim a -unit, so its unit remains in the command budget. Any message object already created for that skipped index is -released. +A deliverable index rejected before handoff as duplicate, compacted-out, before `startMessageId`, selected for +dead-letter handling, or otherwise ineligible does not claim a unit, so that unit remains command-owned. At normal completion or failure, the command path immediately returns all remaining units. If deserialization fails after `D` messages were transferred, those messages eventually return `D`, while the command path returns `P - D`. +The budget has one terminal transition. The first terminal action returns all command-owned units and closes the +budget; later cleanup returns zero. A direct return for a simple non-batched branch is valid only when it atomically +consumes that budget, so it cannot coexist with an outer terminal drain for the same unit. Returning the unit for a +non-batched item before `startMessageId` intentionally corrects the current leak. + Processing more than `P` native messages is a malformed command and follows the protocol-error behavior above. -Once a message claims one unit, the existing per-message lifecycle owns it. Every terminal branch must return the unit -or discard it after the source broker consumer disappears. In particular, stale-epoch discard after asynchronous -handoff closes the prefetch accounting, returns through the retained source state when it is still active, and releases -the message exactly once. +#### Client terminal outcomes + +"Return" below means credit only the retained source incarnation; if that incarnation is obsolete, the return is a +no-op. + +| Branch | Permit outcome | +| --- | --- | +| Invalid explicit `P`, `B`, `ack_set`, or expansion beyond `P` | Return no credit and close the source connection. | +| Checksum or metadata failure | Return range-valid explicit `P`; otherwise use the early compatibility fallback. | +| Decompression failure | Return resolved `P`. | +| Non-batched duplicate or item before `startMessageId` | Consume and return the `P = 1` budget once. | +| Index cleared by the broker's `ack_set` | No return; the index is outside `P`. | +| Batch duplicate, compaction, or start-position skip | Leave the unit command-owned for the single terminal drain. | +| Batched or non-batched item skipped above the dead-letter threshold | Leave the unit command-owned; dead-letter processing retains the message object. | +| Dead-letter candidate at the threshold and still delivered | Transfer one unit to the normal message lifecycle. | +| Expansion fails after `D` successful handoffs | Messages own `D`; command processing returns `P - D`. | +| Handoff throws before acceptance | Restore any reservation and return the unit through the command terminal action. | +| Normal receive, listener delivery, or stale-epoch discard while the source remains active | The accepted message lifecycle returns its one unit once. | + +Permit ownership is independent from pooled-message ownership. A dead-letter candidate can retain a message object +after its permit is returned. A duplicate normally releases its object, but a combined duplicate/dead-letter branch +must first remove it from dead-letter retention or leave release to the retaining owner. Reconnect, replacement, and +close release retained dead-letter objects exactly once. + +Once a successful handoff transfers one unit, the existing per-message lifecycle owns it. Every terminal branch must +return the unit or discard it after the source broker consumer disappears. In particular, stale-epoch discard after +asynchronous handoff closes the prefetch accounting, returns through the retained source state when it is still +active, and releases the message exactly once. The numeric budget does not own networking state. `ConsumerImpl` creates one opaque permit-state object per broker-consumer incarnation; the object is both the local incarnation token and that incarnation's atomic permit @@ -388,10 +443,16 @@ accumulator. Every transferred `MessageImpl` retains that state in addition to i Reconnect or `CommandCloseConsumer` replaces the state even when the pooled `ClientCnx` is unchanged, and never copies the old accumulator into the replacement. +Command processing captures this source state once and carries it through payload, crypto, chunk, queue, listener, and +callback paths. A terminal branch must not reacquire permit state from the current `ClientCnx`, because that same +physical connection can already represent a replacement consumer. + A return can update only its retained state and succeeds only while that state is current. After a threshold drain, the source connection's event loop revalidates exact state identity immediately before writing `CommandFlow`. A new state is not Flow-enabled until its Subscribe succeeds. +Pause can delay writing accumulated credit; resume flushes that existing credit without manufacturing another return. + The connection event loop is the final ordering point for Flow and Subscribe writes: ```text @@ -402,6 +463,30 @@ replacement becomes visible first => old accumulator or queued Flow is discarded Thus a queued old Flow either precedes the replacement Subscribe or becomes a no-op. It cannot enter the replacement accumulator or be written after its Subscribe, and isolation does not require closing the multiplexed connection. +#### Queue, parent-consumer, and zero-queue integration + +Removing accepted messages for redelivery while the same broker consumer remains active returns one unit per removed +message to its retained source state. Clearing a queue during close, reconnect, or replacement releases message +objects but does not transfer old credit to the replacement. A bulk return is valid only when all removed messages +are proven to share the same current source state. + +Multi-topic and partitioned parent consumers do not create new command debt. If a child dequeue already returns a +unit, a parent skip or listener path cannot return it again. If listener mode deliberately defers the child return, +the parent returns it once through the child's retained source state. + +A zero-queue non-batched receive has `P = 1` and returns that unit at its receive or listener terminal point. Native +batch delivery remains unsupported by that specialization; rejection closes the consumer and sends no Flow credit +for the rejected command. A stale zero-queue message is released without crediting a replacement incarnation. + +#### Specialized payload ownership + +Successful decryption that returns to built-in native processing remains covered by the command budget. Once chunk +assembly, a crypto failure action, or a custom `MessagePayloadProcessor` takes ownership, its existing permit +cardinality remains authoritative and the generic budget must not also drain. Existing chunk, crypto `DISCARD`, +`CONSUME`, decrypt-failure-listener, and custom-processor returns remain source-incarnation-bound. Crypto `FAIL` +continues to return no permit while retaining the message for redelivery. A failure before ownership transfers to one +of these paths remains command-owned and follows the normal terminal action. + ## Public-facing Changes ### Binary protocol @@ -428,9 +513,11 @@ The wire rules are: - zero is invalid; no command is sent when nothing is deliverable; and - the value is the actual debit, not necessarily the original batch size. -No protocol-version bump or feature flag is required. Old protobuf readers ignore the optional field, so an upgraded -broker can send it without detecting client support. New Java clients use field presence to distinguish an explicit -debit from an old-broker command that requires the compatibility fallback rules. +No protocol-version bump or `FeatureFlags` capability is added. Old protobuf readers ignore the optional field, so an +upgraded broker can send it without detecting client support. Field absence always invokes the compatibility fallback; +a new client therefore cannot distinguish an old broker, an upgraded broker that omitted the field on one +serialization path, or an intermediary that stripped it. This preserves wire compatibility but intentionally loses +the exact early-failure guarantee for that command. # Monitoring @@ -439,6 +526,9 @@ No metric is added. Existing `pulsar_consumer_available_permits` or but none independently proves that both sides agreed on `P`. Tests therefore verify the accounting invariant directly and verify that repeated corrupted batches do not stall a Shared consumer. +Without a capability flag, monitoring also cannot distinguish an old peer, an upgraded serialization path that +omitted the field, or an intermediary that removed it. + # Security Considerations The broker-to-client field grants no client authority and does not change authentication, authorization, or tenant @@ -486,71 +576,74 @@ rollback. - **Continue inferring from batch metadata:** cannot handle checksum or metadata failure before `B` is parsed and cannot describe a partial-batch debit by itself. - **Use only `ack_set`:** exact for partial batches, but complete batches generally omit it. +- **Add only the wire field:** fixes early ambiguity but does not align post-admission broker aggregates or prevent + client ownership leaks and duplicate returns. - **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. - **Add a reset or absolute-permit command:** requires ordering, idempotency, in-flight command, and connection- generation rules. Attaching the debit to the command that creates it is smaller. +- **Add a protocol capability flag:** distinguishes an old peer from field stripping, but requires handshake and proxy + capability plumbing. This PIP accepts optional-field fallback instead. - **Assign arbitrary permit costs to `Message`:** unnecessary for native batches, where each accepted logical message owns exactly one unit. Deferred processing paths require their own ownership rules. - **Separate broker and client PIPs:** risks defining different debit and return semantics. The selected design keeps - two implementation layers under one end-to-end protocol contract and one pull request. + both layers under one end-to-end protocol contract. # General Notes -The implementation is organized as one end-to-end pull request with two commit layers: - -1. **Protocol and broker:** add the field, finalize one per-entry value, align consumer and persistent Shared - aggregate debits in both modern and classic implementations, serialize that value, make failed writes remove the - affected consumer, and test old-reader compatibility. -2. **Java native batch:** implement resolution and fallbacks, command-local ownership, consumer-incarnation binding, - the narrow terminal fixes, and the persistent Shared end-to-end test. - -Keeping both layers in one PR makes the complete invariant reviewable together; commits retain the same boundary for -review and later backporting. +This PIP defines behavioral boundaries, not required helper classes or commit structure. Existing components can keep +their non-accounting responsibilities, but they must consume the authoritative finalized value, command budget, and +source-incarnation state defined above. The required test matrix covers: -- full and partial native batches, including explicit `P = 1` and old-broker field absence; -- filtering or pending-ack rejection before broker finalization; -- consumer and persistent Shared debits equal the sum of the per-command values in both modern and classic - dispatchers, while each command carries its `P`; +- full and partial native batches, including explicit `P = 1`, every old-broker fallback source, and early failure + before metadata is available; +- invalid explicit zero or unsigned range, `B = 0`, all-zero or out-of-range `ack_set`, inconsistent metadata, and + native expansion beyond `P`; +- old readers, Pulsar proxy forwarding, and a simulated field-stripping intermediary; +- filtering or pending-ack rejection, including `P = 0`, produces no command, debit, or pending-ack ownership; +- consumer debit, every serialized field, and all six listed dispatcher aggregates use the finalized values in + normal, replay, and chunk-specific loops; +- persistent and non-persistent single-active dispatchers debit only the selected consumer and serialize `P`; - batch overshoot from one available permit to a negative broker balance, followed by exact recovery when the client returns the full `P`, in both modern and classic persistent Shared dispatchers; - asynchronous lifetime of finalized per-entry values while existing sender inputs are recycled; -- asynchronous write failure removes the affected consumer from its connection and subscription/dispatcher; -- explicit-field validation and old-broker `ack_set`/metadata fallbacks; +- finalized `avgMessagesPerEntry` with zero emitted entries and both `preciseDispatcherFlowControl` settings; - checksum, metadata, decompression, and mid-batch deserialization failures; -- native skips, stale-epoch discard, different-connection replacement, same-`ClientCnx` recreation, a concurrent - old-return/recreation race, and a queued old Flow that executes after same-connection state replacement; and -- deterministic randomized native batches whose decoded `CommandFlow.messagePermits` sum equals the explicit `P` +- non-batched and batched duplicate, compaction, `startMessageId`, and dead-letter outcomes, including pooled-message + cleanup for combined duplicate/dead-letter state; +- failed handoff, stale-epoch discard, same-incarnation redelivery, reconnect queue clearing, and close; +- synchronous, asynchronous, batch, and listener receive terminal returns; +- multi-topic and partitioned parent forwarding with and without listener-mode deferral; +- zero-queue non-batched delivery, unsupported native batch rejection, and stale old-incarnation cleanup; +- different-connection replacement, same-`ClientCnx` recreation, a concurrent old-return/recreation race, and a + queued old Flow that executes after same-connection state replacement; Subscribe success and pause/resume must not + duplicate or transfer accumulated credit; +- chunk, crypto-failure, and custom processor paths retain existing cardinality without also draining the generic + budget, while successful decryption returning to the built-in path uses exact `P`; any return remains + source-incarnation-bound; +- deterministically generated native batches whose decoded `CommandFlow.messagePermits` sum equals the explicit `P` sum after all delivered messages and immediate skips reach terminal handling; and - repeated corrupted batches on a persistent Shared subscription without dispatch stall. # Appendix A: Implementation map and convergence points -This appendix records where the current behavior is distributed and identifies the two narrow accounting components -introduced by this design. It is supporting detail, not additional implementation scope. +This appendix records where the current behavior is distributed. It is supporting detail, not a required helper-class +layout or additional implementation scope. | Component | Responsibility | Relevance to this PIP | | --- | --- | --- | | `MessageMetadata.num_messages_in_batch` | Stores original payload cardinality `B`. | Payload structure is not always the actual debit and is unavailable before metadata parsing. | | `AbstractBaseDispatcher.filterEntriesForConsumer` | Extracts `B`, `ack_set`, and aggregate counts. | Runs before final pending-ack admission. | | `Consumer.sendMessages` | Performs pending-ack admission and debits consumer/unacked counts. | Must finalize the values for entries that survive admission. | -| PIP-491 `SendMessagesResult` | Carries one primitive per-entry `P` array, its checked sum, and the asynchronous write result for one send. | Establishes the broker finalization boundary consumed by accounting and serialization. | -| `PersistentDispatcherMultipleConsumers` | Default persistent Shared implementation; debits aggregate `totalAvailablePermits`. | Must consume the same finalized aggregate rather than recalculate it. | -| `PersistentDispatcherMultipleConsumersClassic` | Configurable classic persistent Shared implementation; independently debits aggregate `totalAvailablePermits`. | Must consume the same finalized aggregate and preserve the invariant when the rollback setting selects it. | +| Aggregate dispatchers listed above | Debit aggregate available permits after a send. | Must consume the finalized aggregate rather than recalculate it. | | `PulsarCommandSenderImpl.sendMessagesToConsumer` | Serializes surviving entries and `ack_set`. | Must serialize the finalized per-entry value without recomputation. | | `EntryBatchIndexesAcks` | Carries mutable pooled per-entry `ack_set` state. | The sender owns and recycles it asynchronously, so finalized values need an explicit lifetime and must not be recomputed after handoff. | | `ConsumerImpl.messageReceived` | Verifies checksum, parses metadata, decompresses, and chooses the payload path. | Must resolve `P` before payload-dependent failures. | | `ConsumerImpl.receiveIndividualMessagesFromBatch` | Splits native batches and counts skipped messages. | Must transfer units from the command budget instead of reconstructing returns. | | `ConsumerImpl.executeNotifyCallback` | Performs asynchronous callback handling. | A transferred unit needs an exact terminal outcome on its source permit state. | -| PIP-491 `MessagePermitAccounting` | Resolves and validates `P` and owns the numeric command budget. | Centralizes permit inference and conservation without coupling the budget to connection or queue state. | -| PIP-491 local `ConsumerPermitState` | Acts as the identity token and permit accumulator for one broker-consumer incarnation. | Keeps incarnation identity separate from the existing message-source `ClientCnx` and prevents old asynchronous returns from crossing a same-connection consumer recreation. | - -The broker result introduced by this PIP uses one primitive array per send and avoids allocating one object per -dispatched entry. It does not need to be pooled; existing pooled sender inputs retain their current recycling rules. -This low-allocation shape does not change the protocol contract. # Appendix B: Related work outside this PIP @@ -561,11 +654,14 @@ The following known areas have different ordering or ownership requirements and - custom `MessagePayloadProcessor` output cardinality; - encrypted payload and chunk-assembly lifecycles; - non-Java client prefetch implementations; -- multi-topic parent and zero-queue specializations beyond behavior reached through the native child consumer; - broader entry, batch, and logical-message unit consistency in statistics and rate limiting. -These areas remain compatible with the explicit debt and incarnation model but are outside the native Java Shared -invariant defined by this PIP. +These areas remain compatible with the explicit debt and incarnation model, but their own ordering, routing, or +output-cardinality semantics are not redefined by this PIP. The Flow-removal race is tracked by +[issue #26288](https://github.com/apache/pulsar/issues/26288) and +[PR #26289](https://github.com/apache/pulsar/pull/26289). That work touches some of the same counters but enforces a +Flow-side consumer-removal invariant; implementations must account for merge order without combining its scope with +this PIP's send-side command debt. # Links @@ -582,6 +678,9 @@ invariant defined by this PIP. * Current Java paths: [ConsumerImpl](../pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java) and [ConsumerBase](../pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java) -* Accounting convergence types: broker `SendMessagesResult` and Java `MessagePermitAccounting`. +* Adjacent Flow-removal race issue: + https://github.com/apache/pulsar/issues/26288 +* Adjacent Flow-removal race pull request: + https://github.com/apache/pulsar/pull/26289 * Mailing List discussion thread: * Mailing List voting thread: