[fix][broker] Fix persistent throughput degradation caused by permit loss during frequent reconnects on Shared subscriptions

[void-ptr974]

Fixes #26288

Motivation

For Shared subscriptions, the consumer permit counter is updated immediately on the connection EventLoop, while the dispatcher permit counter is updated asynchronously by a task submitted to the broker executor. Consumer removal can run before that queued task.

The following diagram shows one possible failing execution order. Time flows downward.

sequenceDiagram
    participant A as Connection EventLoop (Thread A)
    participant C as Consumer
    participant B as Broker Executor (Thread B)
    participant D as Shared Dispatcher

    Note over A,D: Initial state: removed consumer permits = 10, dispatcher total = 20

    A->>C: t1: handleFlow(+1000)
    C->>C: messagePermits: 10 → 1010
    C-->>B: Queue internalConsumerFlow(+1000)

    Note over B: Flow task has not run yet

    A->>D: t2: handleCloseConsumer() → removeConsumer()
    D->>D: totalAvailablePermits: 20 - 1010 = -990
    D->>D: Remove consumer from consumerSet

    B->>D: t3: Run internalConsumerFlow(+1000)
    D->>D: Consumer is already removed, ignore Flow

    Note over D: Dispatcher total remains -990, but the correct value is 10

Loading

At t2, consumer removal subtracts all 1,010 permits even though the queued Flow task has not added its 1,000 permits to the dispatcher total. At t3, that task cannot restore the count because the consumer has already been removed.

Frequent consumer reconnects can accumulate this negative permit drift and cause persistent consumption throughput degradation.

Modifications

• Track consumer permits that are still pending application to the dispatcher.
• When removing a consumer, subtract only permits already included in the dispatcher total.
• Mark pending Flow permits as processed before checking whether the consumer is still connected.
• Apply the fix to both the default PIP-379 and classic Shared dispatchers.
• Add deterministic permit-accounting tests for both dispatcher implementations and a behavior test with real Shared consumers.

In the example above, removal now subtracts only 1,010 - 1,000 = 10 permits, leaving the dispatcher total at the correct value of 10.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• ./gradlew :pulsar-broker:test --tests org.apache.pulsar.broker.service.persistent.SharedDispatcherPermitAccountingTest
• ./gradlew :pulsar-broker:checkstyleMain :pulsar-broker:checkstyleTest
• ./gradlew quickCheck

The regression test fails on the unpatched code with the dispatcher permit total at -990 instead of 10, and passes with this change.

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

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

The change adds a per-consumer monitor around the two permit counters used by Flow processing and consumer removal. No callbacks or I/O are executed while holding this monitor.