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

[Denovo1998]

PendingDispatcherFlowPermits is now updated for every consumer type, but only the persistent multiple-consumer dispatchers call completePendingDispatcherFlow() or getAvailablePermitsForDispatcherRemoval(). For instance, PersistentDispatcherSingleActiveConsumer and NonPersistentDispatcherMultipleConsumers never drain this counter. As a result, it will keep accumulating for unrelated subscription types, and every Flow command will incur the new monitor overhead.

Could we scope this accounting to the affected persistent Shared or Key_Shared paths—for example, by checking isPersistentTopic && Subscription.isIndividualAckMode(subType)—or expose it as an explicit dispatcher capability?

[void-ptr974]

Thanks, I used the suggested scope. Pending accounting now runs only for persistent Shared and Key_Shared consumers, while all other consumer types keep the original lock-free Flow path. I also added coverage for the untracked subscription types.

[Denovo1998]

This accounting balance can legitimately become negative, and this behavior is important to preserve explicitly. While a Flow update is still pending, dispatch can consume newly visible messagePermits before the queued internalConsumerFlow has added those permits to totalAvailablePermits. In that scenario, messagePermits — pendingDispatcherFlowPermits can be negative. Subtracting that negative balance during removal is necessary to correct the dispatcher total.

Could we document this invariant and add a regression test covering: “pending Flow → dispatch some newly visible permits → remove before the Flow task executes”? The current no‑op dispatcher tests do not exercise this case. Otherwise, a future cleanup like Math.max(0, …) might seem reasonable but would reintroduce permit drift.

[void-ptr974]

Good point. The Javadoc now states that the removal balance may legitimately be negative and must not be clamped. I also added a regression test with an in-flight Flow, actual message dispatch, and two consumers remaining after removal.

[Denovo1998]

The production change affects both Key_Shared implementations. The ModernPersistentStickyKeyDispatcherMultipleConsumers inherits the modern dispatcher, while the classic sticky dispatcher derives from the classic one. Entry-bucket dispatch follows the modern sticky path. Current tests only create SubType.Shared consumers.

Would it be valuable to add at least modern and classic Key_Shared variants for the Flow.remove race, given that their removal path wraps super.removeConsumer() with selector and hash-draining state changes?

[void-ptr974]

Added a four-way test matrix covering Shared and Key_Shared on both the modern and classic dispatcher implementations.

[lhotari]

Reviewed this in depth and verified it against a local build. The core fix is correct - absent integer overflow (see the inline note on the Math.max clamp) I could not find a case where it makes accounting worse. Everything below is non-blocking, on top of @Denovo1998's review, which I think is asking the right questions.

What I verified

The diagnosis is right, and the failure is stronger than a narrow race. ServerCnx.handleFlow runs on the connection event loop and defers the dispatcher update to the broker executor, while handleCloseConsumer -> Consumer.close -> PersistentSubscription.removeConsumer -> dispatcher.removeConsumer all run synchronously on that same event loop. Any close shortly after a Flow hits this while the queued task is still pending - exactly the reconnect workload in #26288. The unload path is covered too, since disconnectAllConsumers holds the dispatcher monitor across consumer.disconnect() -> removal.

The symptom matches the issue. readMoreEntries floors the read at Math.max(totalAvailablePermits, getFirstAvailableConsumerPermits()), so a negative total does not stall the subscription - with several consumers connected it shrinks each read toward a single consumer's permits instead of their sum. That is the "30-40% slower but still progressing" behaviour in #26288 rather than a hard stop.

The invariant restored is exact: totalAvailablePermits == sum(messagePermits - pendingDispatcherFlowPermits) over connected consumers. Checked at every mutation site: both internalConsumerFlow branches, Consumer.sendMessages against the matching TOTAL_AVAILABLE_PERMITS_UPDATER decrements in both dispatchers and the sticky-key one, all three removeConsumer branches, clearComponentsAfterRemovedAllConsumers, and the blocked-permits path (PERMITS_RECEIVED_WHILE_CONSUMER_BLOCKED correctly sits in neither side). The accounting is also commutative: BrokerService.executor() round-robins Flow tasks across threads so they can complete out of order, but each task completes exactly what it added, so every interleaving converges.

Threading is sound. flowPermitAccountingLock is a leaf lock - nothing is acquired under it and no callback or I/O runs inside - so there is no ordering relationship with the dispatcher monitor to get wrong, and the new critical sections are two field updates.

On @Denovo1998's per-Flow overhead concern: it is a per-Consumer monitor, contended only between that consumer's event loop and the broker executor. flowPermits already calls System.currentTimeMillis() and submits a task to an executor in the same method, so the uncontended monitor is small next to what is already there. I do not think overhead alone justifies restructuring, though the correctness argument for scoping still stands.

Executor rejection is correct by construction - I went looking for a leak and there isn't one. If consumerFlow is ever rejected at shutdown the flow legitimately stays pending, and removal then subtracts exactly the permits that were applied.

Tests pin the fix. On master with only SharedDispatcherPermitAccountingTest added, all three methods fail with expected: 10 but was: -990; on this branch all 5 invocations pass in 3.3s. :pulsar-broker:checkstyleMain :pulsar-broker:checkstyleTest pass.

On the open review points

• The counter-scoping question (Consumer.java:931) is real, and I have added inline the consequence that I think makes it worth acting on: the accessor becomes an active trap for whoever extends this fix to the other dispatchers.
• The legitimately-negative balance (Consumer.java:1000) is real and load-bearing. I have posted the mechanism and a test shape inline. Note it needs no batching or special configuration - it follows from dispatch being sized off getAvailablePermits(), which includes permits the dispatcher has not counted yet.
• On Key_Shared coverage, agreed - the three implementations are listed inline. Their removeConsumer overrides wrap super.removeConsumer with selector and draining-hash work, so a Key_Shared variant of the race test is worth having even though the permit arithmetic itself is inherited unchanged.

Out of scope, but worth not losing

The same defect class survives on non-persistent Shared subscriptions: NonPersistentDispatcherMultipleConsumers.removeConsumer still subtracts the full getAvailablePermits(), and its consumerFlow drops updates from consumers already out of consumerSet. The window is narrower because that consumerFlow is synchronous, but not empty - NonPersistentTopic.onPoliciesUpdate -> Consumer.checkPermissionsAsync -> disconnect() -> close() runs removal on the authorization future's completion thread rather than the connection event loop, and disconnectAllConsumers holds the dispatcher monitor across the whole teardown. There the consequence is worse than a slowdown: sendMessages drops entries when the total is not positive. Full teardowns self-heal, since the total is reset to 0 once the consumer list empties, so lasting drift needs a partial removal that leaves survivors.

I would keep that out of this PR - it needs the scoping question settled first - but it is worth a follow-up issue.

Backport note

The release/4.2.5 and release/4.0.14 labels are on this PR: the diff touches slog-style logging (log.debug().attr(...)), and branch-4.2 / branch-4.0 are Maven + slf4j, so those cherry-picks will need the usual logging adaptation rather than a clean pick.

[lhotari]

This clamp is on the pending counter, a different quantity from the messagePermits - pendingDispatcherFlowPermits balance @Denovo1998 flags below at line 1000. That balance must stay signed and I agree with them there; this is a narrower point about the counter itself.

For the two dispatchers that call completePendingDispatcherFlow today the add/complete pairing is exact, so the negative branch is unreachable absent integer wrap. It is reachable through wrap: ServerCnx.handleFlow forwards flow.getMessagePermits() with no upper bound and flowPermits only checks > 0. Three Flow(Integer.MAX_VALUE) commands whose dispatcher tasks are all still queued leave pendingDispatcherFlowPermits at 2147483645, and the first completePendingDispatcherFlow to run then computes -2, silently clamped to 0.

Why the clamp is the wrong tool: every mutation of these two fields on the broker's own paths is a delta (the only absolute writes are the two constructors and updateStats, which nothing in the broker calls), and Java int arithmetic is exactly mod 2^32. So without the clamp, messagePermits - pendingDispatcherFlowPermits stays congruent mod 2^32 to (permits applied to the dispatcher total) minus (permits consumed by dispatch, net of the batch-index-ack credit at Consumer.java:446) - the accounting survives the wrap intact. Math.max is the one non-delta operation here, and it is what breaks that congruence. The discarded offset then persists into every later add, so from that point on the removal value no longer tracks the unclamped one.

To be clear, this is not a new externally-triggerable vulnerability: messagePermits and totalAvailablePermits are plain ints on master and already wrap on the same input. What the clamp does is discard information. (The missing upper bound on flow.getMessagePermits() is a pre-existing hardening gap deserving its own issue, not something to fix here.)

Suggestion: log or count the negative branch rather than silently absorbing it, so the signal survives if a future dispatcher or a backport ever introduces a genuine add/complete mismatch - exactly the failure mode this PR exists to prevent.

[void-ptr974]

Removed the clamp, so completion now applies the raw int delta and preserves the existing modulo-2^32 accounting behavior. I also added a regression test with three queued Flow(Integer.MAX_VALUE) updates. I did not add logging because a negative pending value cannot reliably distinguish signed wrap from an add/complete mismatch. Flow upper-bound hardening remains separate from this fix.

[lhotari]

Building on @Denovo1998's point that this counter is not drained for every subscription type, there is a sharper consequence worth capturing in this javadoc: this accessor returns a meaningless value for any dispatcher that never calls completePendingDispatcherFlow.

PersistentDispatcherSingleActiveConsumer.consumerFlow drops the permit count entirely, NonPersistentDispatcherMultipleConsumers.consumerFlow applies permits synchronously without completing, and NonPersistentDispatcherSingleActiveConsumer.consumerFlow is a no-op. For those consumers pendingDispatcherFlowPermits only ever grows, so messagePermits - pendingDispatcherFlowPermits degrades into minus the net number of messages ever dispatched to that consumer - 0 for a consumer that has flowed but not yet received anything, and increasingly negative over its lifetime.

That makes this a trap for the obvious follow-up. NonPersistentDispatcherMultipleConsumers.removeConsumer still performs the same subtraction this PR is fixing (TOTAL_AVAILABLE_PERMITS_UPDATER.addAndGet(this, -consumer.getAvailablePermits()), lines 117 and 121). Swapping in getAvailablePermitsForDispatcherRemoval() there would give totalAvailablePermits -= (large negative), crediting the dispatcher with every message that consumer was ever dispatched - a considerably worse bug than the one being fixed.

However the scoping question is resolved, could the precondition be stated here explicitly, e.g. "only valid on dispatchers that call completePendingDispatcherFlow"?

On the scoping itself: the suggested isPersistentTopic && Subscription.isIndividualAckMode(subType) gate is exact today - isIndividualAckMode is Shared || Key_Shared, and those subscriptions always land on the two patched dispatchers or a subclass of them (PersistentStickyKeyDispatcherMultipleConsumers, its ...Classic counterpart, and the PIP-486 PersistentEntryBucketDispatcherMultipleConsumers all inherit consumerFlow unchanged and call super.removeConsumer). My only reservation is that it encodes "which dispatchers complete pending" in a second place that can drift; letting the dispatcher drive the tracking would be drift-proof, at the cost of a wider diff.

[void-ptr974]

Added the explicit precondition to the Javadoc: this accessor is only valid for persistent Shared and Key_Shared dispatchers whose Flow tasks complete the pending accounting. The new scope guard ensures other consumer types do not accumulate this counter.

[lhotari]

Corroborating @Denovo1998's request for a regression test on the legitimately-negative balance: I traced it, and it is both real and load-bearing.

messagePermits - pendingDispatcherFlowPermits goes negative whenever a consumer is dispatched more messages than the portion of its permits the dispatcher has actually counted. The mechanism is that trySendMessagesToConsumers sizes each consumer's share from c.getAvailablePermits(), i.e. messagePermits - which already includes permits from a Flow whose dispatcher task is still queued and therefore not yet in totalAvailablePermits. So any dispatch while a Flow is in flight can push the balance below zero, and subtracting that negative balance on removal is exactly what restores the invariant. No batching or unusual configuration is needed; the tests in this PR reproduce it with no ledger read at all.

A robust shape for the regression test: give the consumer being removed a large in-flight Flow, let a dispatch consume more than messagePermits - pending, remove it before the Flow task runs, and assert the dispatcher total equals the sum of the remaining consumers' messagePermits. Worth leaving two consumers connected after the removal - with only one left, the Math.max(totalAvailablePermits, getFirstAvailableConsumerPermits()) floor in readMoreEntries masks a wrong total.

The invariant is worth stating near here too:

totalAvailablePermits == sum over connected consumers of (messagePermits - pendingDispatcherFlowPermits)

The Flow path never perturbs it, since addPermitsPendingDispatcherUpdate raises both fields by the same amount. The only transient violation is inside trySendMessagesToConsumers, between the messagePermits decrement at Consumer.java:446 and the matching TOTAL_AVAILABLE_PERMITS_UPDATER decrement - both inside the dispatcher's synchronized send path. Removal reads the balance under that same monitor, so it is consistent exactly where it is used.

[void-ptr974]

Added the invariant comments at both dispatcher removal sites and the suggested regression test with two surviving consumers. The test verifies that the dispatcher total equals the sum of their available permits after removal.

[lhotari]

This block takes the dispatcher monitor and then, through removedBrokerConsumer.close(), the subscription monitor: Consumer.close -> PersistentSubscription.removeConsumer (synchronized) -> dispatcher.removeConsumer.

Production nests them the other way, and the order is documented in code - see the // Lock the Subscription object before locking the Dispatcher object to avoid deadlocks comment in PersistentSubscription above synchronized (this) { ... dispatcher.disconnectActiveConsumers(true); }. PersistentSubscription.disconnect, close and addConsumerInternal all follow that order.

I could not find a subscription-then-dispatcher path that actually fires during this short window in this setup, so the practical risk looks low - but those paths are reachable from broker background work (inactive-subscription checks, topic GC, unload), and if one ever lands here it surfaces as the 30s timeOut rather than a clear failure, which is an expensive kind of CI flake to chase. Acquiring the subscription monitor first would remove the inversion without changing what the test exercises.

One smaller observation, not an ask: the broker-side close() here is followed by removedClient.close() on line 157. Consumer.close does call cnx.removedConsumer(this), but ServerCnx.safelyRemoveConsumer only schedules the map removal onto ctx.executor() via whenCompleteAsync. Both that task and the inbound CloseConsumer run on the same channel event loop with the removal merely enqueued first, so it normally wins, but it is not guaranteed. In the losing interleaving the second close reaches the defensive else branch in removeConsumer and logs Trying to remove a non-connected consumer at ERROR (assertions still hold, since that branch leaves totalAvailablePermits alone as long as a consumer remains, which is the case here). The broker-side close inside the monitor is what makes the removal deterministic, so I would keep it - a one-line comment noting the possible ERROR would save the next person chasing it.

[void-ptr974]

Updated the integration test to follow the production subscription-then-dispatcher lock order. It now also waits for the broker-side connection-map cleanup before closing the client, avoiding a duplicate dispatcher removal and its error log.

[lhotari]

This test and the next one reproduce the race by holding the dispatcher's intrinsic monitor so the queued internalConsumerFlow cannot run before the removal. That works because internalConsumerFlow is synchronized on the dispatcher instance, so the queued task blocks until the test thread - which performs the removal itself - releases the monitor. Copying the comment the third test already has at line 149 onto these two blocks would help the next reader.

I verified these genuinely pin the fix: on master with only this test file added, all three methods fail with expected: 10 but was: -990, and they pass on this branch.

The risk is silent decay rather than breakage. If dispatcher flow processing ever moves off the intrinsic monitor, the ordering becomes a timing race instead of a guarantee - and because the final totals under this fix are order-independent, the assertions stay green while no longer forcing the interleaving they were written for.

A cheap guard: assert inside the synchronized block that totalAvailablePermits is still the pre-Flow value. The third test asserts a consumer-side counter inside its block, but that one cannot detect the decay - flowPermits bumps messagePermits synchronously on the test thread either way, so only the dispatcher total distinguishes "Flow task still queued" from "Flow task already applied".

[void-ptr974]

Added comments explaining the monitor-based ordering and assertions inside the synchronized blocks that the dispatcher total is still unchanged. This keeps the intended queued-Flow interleaving explicitly covered by the tests.

[void-ptr974]

@lhotari Thanks for the detailed review and the follow-up suggestions.

I have pushed e2467d03493 with the inline feedback incorporated. Pending tracking is now limited to the affected subscription types, the signed and wrapped accounting behavior is preserved, and the regression coverage includes the negative-balance case and the Shared/Key_Shared dispatcher variants. All 14 test invocations and the broker checkstyle checks pass.

I agree that the non-persistent Shared case should be handled separately. I will first check whether it is related to #24018 before opening a new issue.

I have also noted the backport point. Since the release labels are already on the PR, no further action is needed here, and I can help with branch-specific logging adaptation if needed.

[Denovo1998]

LGTM.