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

[void-ptr974]

Fixes #26288

Motivation

For persistent Shared and Key_Shared subscriptions, a Flow command updates the Consumer permit counter before the corresponding update is applied to the dispatcher total. Previously, consumer removal could run while that dispatcher update was still queued.

The following diagram shows the pre-fix failure. Time flows downward.

sequenceDiagram
    participant IO as Connection EventLoop
    participant C as Consumer
    participant W as Queued Flow worker
    participant D as Shared Dispatcher

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

    IO->>C: handleFlow(+1000)
    C->>C: messagePermits: 10 -> 1010
    C-->>W: Queue dispatcher update (+1000)

    Note over W: Flow task has not run yet

    IO->>D: removeConsumer(C)
    D->>D: Old code: totalAvailablePermits = 20 - 1010 = -990
    D->>D: Remove C

    W->>D: Run queued Flow update
    D->>D: C is already removed, ignore update

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

Loading

This negative permit drift does not necessarily stop consumption. It reduces the effective permits used to size reads and can cause persistent throughput degradation after frequent reconnects.

Modifications

• Track Consumer permits that are pending application to the dispatcher.
• During removal, subtract only the represented balance: messagePermits - pendingDispatcherFlowPermits.
• Route persistent Shared/Key_Shared Flow processing through the dispatcher's existing dispatchMessagesThread in both modern and classic implementations.
• Complete pending accounting and update totalAvailablePermits under the dispatcher monitor, then initiate the read from the same dispatcher lane.
• Match queued Flow updates by exact Consumer instance so an old task cannot apply permits to an equal replacement Consumer after reconnect.
• If lane submission is rejected during shutdown, leave the update pending; removal will continue to exclude those unapplied permits.
• Add deterministic coverage for accounting, EventLoop progress, lane affinity, removal/replacement races, signed permit balances, and executor rejection.

No executor or thread pool is added.

Final threading model

sequenceDiagram
    participant IO as Connection EventLoop
    participant C as Consumer
    participant L as dispatchMessagesThread (topic-worker lane)
    participant D as Shared Dispatcher

    IO->>C: handleFlow(+N)
    C->>C: Under accounting lock:<br/>messagePermits += N<br/>pendingDispatcherFlowPermits += N
    C-->>L: Enqueue Flow task
    Note over IO,L: EventLoop returns without acquiring<br/>or waiting for the dispatcher monitor

    opt Consumer is removed before the queued task
        IO->>D: removeConsumer(C)
        activate D
        D->>C: Under dispatcher then accounting lock:<br/>represented balance = messagePermits - pending permits
        D->>D: Subtract balance and remove C
        deactivate D
    end

    L->>D: Process queued Flow
    activate D
    D->>C: Under dispatcher then accounting lock:<br/>pendingDispatcherFlowPermits -= N
    D->>D: Check exact Consumer-instance membership
    alt Original Consumer is still connected
        D->>D: totalAvailablePermits += N
    else Consumer was removed or replaced
        D->>D: Do not change dispatcher total
    end
    deactivate D

    opt Original Consumer is still connected
        L->>D: readMoreEntries() on the same lane
    end

Loading

The connection EventLoop performs only the Consumer-side accounting and lane submission. Dispatcher-monitor waiting, dispatcher-total accounting, and the Flow-triggered read occur on the selected topic-worker lane.

The modern Flow path intentionally initiates the read directly from that lane. Reusing the existing readMoreEntriesAsync() would submit the read back to BrokerService.executor() (the Netty I/O worker group) and could reintroduce dispatcher-monitor waiting on an I/O thread. Existing dispatcher guards prevent duplicate cursor reads; lane-local trigger coalescing can be considered separately as a CPU optimization.

Under the dispatcher monitor, the maintained invariant is:

totalAvailablePermits = sum(messagePermits - pendingDispatcherFlowPermits)

The arithmetic intentionally preserves the existing signed Java int wrap behavior, and the removal balance may legitimately be negative when dispatch consumes permits before a queued Flow update is applied.

Verifying this change

• Make sure that the change passes the CI checks.

Targeted validation completed locally and on the dedicated stress host:

• SharedDispatcherPermitAccountingTest: 19/19 passed.
• SharedDispatcherFlowThreadingTest: 18/18 passed.
• Existing modern/classic Shared and Key_Shared dispatcher regression classes: 36/36 passed.
• ./gradlew :pulsar-broker:checkstyleMain :pulsar-broker:checkstyleTest: passed.
• Mixed modern/classic Shared stress, with dispatcherDispatchMessagesInSubscriptionThread enabled and disabled: all unique messages received, zero duplicates, zero final backlog, and no permit drift.
• Across two matched JFR legs, the inline version recorded 4,748 Flow-path dispatcher-monitor contentions on pulsar-io (1,006 ms total); this version recorded none for that path on pulsar-io, with the work moved to broker-topic-workers.
• Boundary receiverQueueSize=1 stress, producing approximately one Flow per delivered message: end-to-end throughput was unchanged, receive-gap p99 differed by approximately 0.43%, and broker CPU was approximately 12.1% higher. All correctness/progress checks passed. In the q10 control, throughput and p99 remained stable and the broker CPU difference was approximately 3%.

The tests can be run with:

./gradlew :pulsar-broker:test \
  --tests org.apache.pulsar.broker.service.persistent.SharedDispatcherPermitAccountingTest \
  --tests org.apache.pulsar.broker.service.persistent.SharedDispatcherFlowThreadingTest

./gradlew :pulsar-broker:checkstyleMain :pulsar-broker:checkstyleTest

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 short per-Consumer accounting critical section for persistent Shared/Key_Shared subscriptions. No callback, dispatcher monitor, or I/O operation is executed while holding that accounting lock. No user-facing client/admin API, wire protocol, persisted schema, configuration/default, or metric schema is changed; the added Consumer helpers are broker-internal.

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

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

[lhotari]

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.

Waiting for @merlimat's review

[lhotari]

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.

We discussed this with @merlimat and took a look together.
It seems that #16304 is the change that introduced the bug. The fix would be change this code:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

Lines 301 to 324 in 3529aba

	@Override
	public void consumerFlow(Consumer consumer, int additionalNumberOfMessages) {
	topic.getBrokerService().executor().execute(() -> {
	internalConsumerFlow(consumer, additionalNumberOfMessages);
	});
	}
	private synchronized void internalConsumerFlow(Consumer consumer, int additionalNumberOfMessages) {
	if (!consumerSet.contains(consumer)) {
	log.debug()
	.attr("consumer", consumer)
	.log("Ignoring flow control from disconnected consumer");
	return;
	}
	totalAvailablePermits += additionalNumberOfMessages;
	log.debug()
	.attr("consumer", consumer)
	.attr("totalAvailablePermits", totalAvailablePermits)
	.attr("additionalNumberOfMessages", additionalNumberOfMessages)
	.log("Trigger new read after receiving flow control message");
	readMoreEntriesAsync();
	}

In PersistentDispatcherMultipleConsumers, readMoreEntries has been replaced by readMoreEntriesAsync. Calling the internalConsumerFlow in a separate executor should be removed in consumerFlow method and the internalConsumerFlow should be inlined. A similar fix would be needed for PersistentDispatcherMultipleConsumersClassic, although it should also make the switch from readMoreEntries to readMoreEntriesAsync. One important optimization in PersistentDispatcherMultipleConsumers is that readMoreEntriesAsync calls are de-duplicated. Adding such optimizations to PersistentDispatcherMultipleConsumersClassic would be out of scope of this PR. There shouldn't be a need to even use PersistentDispatcherMultipleConsumersClassic. It was added during PIP-379 to give an option to revert to use the previous implementations of the dispatchers in the case of the changes made for Pulsar 4.0 & PIP-379 would cause regressions. A few regressions were found afterwards and they have been fixed.

[void-ptr974]

@lhotari Thanks for the suggestion and for looking into this with @merlimat.

Updated in cf57407fe13.

• Removed the additional executor handoff and inlined Flow accounting into consumerFlow for both modern and classic Shared dispatchers.
• Limited the synchronized section to pending-flow settlement, consumer membership validation, and permit accounting.
• Moved logging and readMoreEntriesAsync() outside the dispatcher monitor.
• Kept the existing de-duplicated async read scheduling for the modern dispatcher.
• Switched the classic dispatcher from readMoreEntries() to readMoreEntriesAsync() without adding de-duplication.
• Retained pending-flow accounting to cover cross-thread consumer removal while a Flow caller is waiting for the dispatcher monitor.
• Updated the behavior tests for modern/classic and Shared/Key_Shared Flow/removal interleavings, including continued delivery after consumer removal.

[void-ptr974]

Added follow-up commit ecdb4b52fed on top of cf57407fe13.

With cf57407, Flow accounting runs inline on the connection EventLoop. Although the read trigger is outside the synchronized block, the EventLoop can still wait for the persistent Shared dispatcher monitor while dispatch/filter/send work owns it.

The follow-up changes the threading model to:

• the connection EventLoop updates the Consumer permits/pending value, submits the Flow task, and returns;
• the existing dispatchMessagesThread completes pending accounting, checks membership, and updates the dispatcher total under the dispatcher monitor;
• read initiation runs from the same dispatcher lane;
• rejected tasks remain pending instead of being caller-run on the EventLoop, so removal still excludes unapplied permits correctly.

No new executor or thread pool is introduced.

Validation:

• 33/33 committed test invocations passed, covering modern/classic and Shared/Key_Shared;
• deterministic tests cover EventLoop progress, dispatcher-lane affinity, queued Flow/removal ordering, negative permit balances, blocked Consumers, and executor rejection;
• additional mixed stress covered modern/classic Shared dispatchers with dispatcherDispatchMessagesInSubscriptionThread enabled and disabled: all unique message IDs were received, with no duplicates, zero final backlog, and no permit drift;
• in a matched JFR run, cf57407 recorded 4,748 Flow-related dispatcher-monitor contentions on pulsar-io, totaling 1,006 ms. The follow-up recorded zero Flow-related dispatcher-monitor contention on pulsar-io; the work moved to broker-topic-workers.

Receiver queue size 1 was used as an intentional boundary stress case, generating approximately one Flow per delivered message. Throughput remained stable, receive-gap p99 differed by less than approximately 0.5%, and all correctness/progress checks passed. In the q10 control, the measured CPU difference was approximately 3%.

The result is that Flow processing no longer makes the connection EventLoop wait for the dispatcher monitor, without an observed correctness, progress, or sub-saturation throughput/latency regression.

[Denovo1998]

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.

[Denovo1998]

PIP-379 intentionally changed the path from readMoreEntries() to readMoreEntriesAsync() and added readMoreEntriesAsyncRequested to deduplicate read triggers. As a result, Flow accounting is now correctly serialized on the dispatchMessagesThread, but each queued Flow re-invokes the full readMoreEntries() path directly, bypassing the deduplication.

The q1/JFR stress results appear acceptable, so this does not seem to be a correctness issue. However, should we either preserve per-dispatch-lane read-trigger deduplication or add a comment clarifying that bypassing the PIP-379 dedup is intentional? Without that, this appears to be an unintended regression of the optimization and may be reverted later.