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

[void-ptr974]

PIP: #26336

Motivation

For every delivered CommandMessage, the broker uses a number of consumer permits and the client must eventually return exactly that number. For a native batch, this count (P) is the number of logical messages that are actually deliverable, which can differ from the original batch size during partial redelivery.

Pulsar does not currently carry P on the wire. Broker layers and the Java client derive related counts independently from batch metadata and ack_set. These calculations normally agree, but can diverge when final admission changes the send set, payload processing fails before or during batch expansion, asynchronous work has no terminal outcome, or a broker consumer is recreated while its pooled connection remains active. Returning too few permits progressively reduces receiver capacity and can stall Shared delivery; returning stale or excess permits weakens backpressure.

Modifications

This PR implements the PIP-491 contract end to end:

• Add optional CommandMessage.message_permits = 6 without a protobuf default. An upgraded broker sends the field for every delivered command, including P = 1; field absence is reserved for old-broker compatibility.
• Finalize one positive P per entry in Consumer.sendMessages after pending-ack admission. Entries rejected by filtering/admission or with no deliverable indexes are neither sent nor debited.
• Keep the per-entry values and their sum in SendMessagesResult. Consumer accounting, both persistent Shared dispatcher implementations, and command serialization all use that same finalized result, independently of recyclable sender inputs.
• Remove the affected broker consumer when an asynchronous message write fails, so failed writes have a terminal accounting outcome.
• Resolve and validate explicit permits in the Java native-message path. A new client falls back to ack_set, parsed batch metadata, and finally the legacy value of one when the field is absent.
• Treat P as a command-local budget during native batch expansion. Successfully handed-off messages claim one unit; skips and unexpanded units remain with the command; the two paths return exactly P without double-returning.
• Scope returned credit to a local ConsumerPermitState representing one broker-consumer incarnation. Reconnect or consumer recreation replaces the state even when the same ClientCnx is reused. The atomic accumulator and connection event loop revalidate state identity before writing CommandFlow, preventing delayed credit from crossing into a replacement consumer.
• Close the source connection without returning credit when an explicit permit value is invalid or inconsistent with native payload metadata and ack_set.

The initial exact guarantee covers persistent Shared delivery (both default and classic implementations) and the Java native-message path. Custom MessagePayloadProcessor output, encrypted/chunked processing, non-Java clients, Key_Shared-specific draining, absolute permit reset/synchronization, and broad dispatcher refactoring remain out of scope.

Compatibility

• New broker + new Java client: explicit end-to-end permit conservation is enforced.
• New broker + old client: the unknown optional field is ignored; existing normal delivery behavior is unchanged.
• Old broker + new Java client: the client uses the existing inference fallbacks.
• Proxies continue to forward the optional field as part of CommandMessage; a proxy that drops unknown fields remains compatible through fallback behavior.

No Java public API, configuration, acknowledgment semantic, receiver-queue semantic, or protocol-version negotiation is changed.

Verifying this change

Local validation on the final source tree:

• ./gradlew quickCheck
• ./gradlew :pulsar-broker:test --tests org.apache.pulsar.broker.service.ConsumerTest :pulsar-broker:checkstyleTest

Independent validation on home3 (Linux, JDK 21) from an exact source archive of commit bd4a56494f3:

• ./gradlew quickCheck: 421 actionable tasks, build successful
• ./gradlew sanityCheck: 584 actionable tasks, build successful
• 55 targeted tests across CommandsTest, ConsumerImplTest, MessagePermitAccountingTest, ConsumerTest, EntryBatchIndexesAcksTest, BatchMessageIndexAckTest, CompactedOutBatchMessageTest, and CorruptedBatchMessagePermitTest: 1 skipped, 0 failures, 0 errors

The targeted matrix covers explicit P = 1, legacy field absence, full and partial batches, bounded ack_set cardinality, filtering and pending-ack rejection, both Shared dispatcher implementations, finalized-value lifetime across recycled sender inputs, write-failure removal, checksum/metadata/decompression/mid-batch failures, native skips, stale epochs, different- and same-ClientCnx replacement, concurrent returns, queued stale Flow, and repeated corrupted batches without a delivery stall.

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

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol: adds backward-compatible optional CommandMessage.message_permits = 6
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment