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

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.

LGTM.

The follow-up commit addresses all five review threads. The pending counter now preserves signed wrap, tracking is restricted to persistent Shared/Key_Shared consumers, the accessor documents its precondition and signed invariant, and the tests cover wrapped and negative balances, both dispatcher variants, queued-Flow ordering, and production lock order. I found no new issue in this review-feedback-only delta.

The PR description has become outdated following the latest threading changes. The original race diagram remains useful for illustrating the pre-fix behavior, but the Modifications and Verification sections no longer describe the current implementation. Changes include: flow updates now run on the dispatchMessagesThread, executor rejection intentionally leaves credits pending, and queued updates use exact Consumer-instance membership to avoid applying stale credit to a replacement Consumer with equal attributes. A new SharedDispatcherFlowThreadingTest has also been added, along with JFR and mixed-stress validation.

Since this PR changes the threading model and is intended for backports, please refresh the description to reflect the final design before merging.

The three new commits move Flow processing off the connection EventLoop instead of inlining it there, which is a different design from the one discussed earlier on this PR. The reasoning and the JFR numbers for that reversal are convincing on their own terms, but it is a broker-wide threading change riding on a permit-accounting fix, so it deserves an explicit decision rather than arriving as a follow-up commit. I did not find a correctness defect in the accounting itself.

What I checked and found sound at f81fcae0:

• The permit invariant is still exact. Consumer.flowPermits raises messagePermits and pendingDispatcherFlowPermits together under flowPermitAccountingLock (service/Consumer.java:974-982), so the removal balance is unchanged until the dispatcher task runs; the task settles the pending value and credits totalAvailablePermits under the dispatcher monitor, for the exact connected instance (PersistentDispatcherMultipleConsumers.java:324-333); removal subtracts the resulting balance (PersistentDispatcherMultipleConsumers.java:262-265). No path holds flowPermitAccountingLock while acquiring the dispatcher or subscription monitor, so the ordering is one-way.
• Leaving the permits pending on RejectedExecutionException is the right call. The broker-topic-workers executor is built with no queue limit (BrokerService.java:389-391), so it rejects only once it has been shut down during broker shutdown, not under load.
• The identity membership check is load-bearing, not merely defensive. Consumer.equals is consumerId plus cnx.clientAddress() (service/Consumer.java:1184-1193), and in the replacement race the set holds the new instance, so contains() would have let a stale Flow task credit permits for a consumer that had already been removed. testQueuedFlowDoesNotApplyToEqualReplacementConsumer pins exactly that.
• No new recursion into readMoreEntries(). The Flow call is a top-level dispatch-lane task, the modern dispatcher already calls readMoreEntries() synchronously from its send/read loop on the same lane, and the recursion-sensitive branches still use readMoreEntriesAsync().
• All five earlier threads still hold at this head, including the two whose test mechanism changed: testRemainingConsumerCanContinueAfterFlowAndCloseRace still follows the subscription-then-dispatcher order and still waits for the connection-map removal, and the queued-Flow ordering is now forced with a dispatch-lane barrier rather than by holding the dispatcher monitor, which is strictly safer now that the monitor no longer gates the Flow task.

The consequences of the new placement, and the shape I would like the read triggering to take, are in the inline comments below.

Scope and backport. The labels put this on 4.0.14 and 4.2.5. The accounting fix is contained -- Consumer bookkeeping plus the two removeConsumer call sites -- while the new commits change which executor every persistent Shared and Key_Shared consumerFlow runs on, and how reads are triggered for all of them. Would you consider splitting the threading change into its own PR, so what lands on the maintenance branches is only the accounting fix?

The accounting-only revision addresses my remaining scope and read-trigger concerns. The signed pending-permit accounting and regression coverage are retained. I found no further blocking issue in d72e6c7a9c8.