[fix][broker] Debit un-acked messages only when the consumer is actually removed

[FlorentinDUBOIS]

Relates to #26416. Carved out of #26417 following the review discussion there: the available-permits half of that PR is superseded by #26289, while this double-debit of the un-acknowledged message counter is an independent defect that neither master nor #26289 addresses, so it gets its own PR, review, and backport decision.

Motivation

PersistentDispatcherMultipleConsumers#removeConsumer debits the subscription's un-acknowledged message count by the departing consumer's own count before it establishes whether that consumer is still registered:

addUnAckedMessages(-consumer.getUnackedMessages());
if (consumerSet.removeAll(consumer) == 1) {

The else branch below is the defensive path added by #22270 for a consumer that is still in consumerList but no longer in consumerSet, so that the topic can still be unloaded. Reaching it means the consumer was already removed once, and the first removal already debited its un-acknowledged messages. The unguarded debit therefore subtracts them a second time and drives totalUnackedMessages negative.

That counter is what maxUnackedMessagesOnSubscription throttles on, and nothing resets it while the dispatcher lives — clearComponentsAfterRemovedAllConsumers() resets the available-permits aggregate but deliberately leaves it alone. A negative value therefore silently raises the effective limit for the lifetime of the dispatcher, and since addUnAckedMessages also feeds the broker-wide counter, the drift is not confined to one subscription.

PersistentDispatcherMultipleConsumersClassic#removeConsumer — the documented PIP-379 rollback path, selectable at runtime through the dynamic subscriptionSharedUseClassicPersistentImplementation flag — carries the identical unguarded debit.

Modifications

• Move addUnAckedMessages(-consumer.getUnackedMessages()) inside the consumerSet.removeAll(consumer) == 1 guard in PersistentDispatcherMultipleConsumers, so that only the removal which actually unregisters the consumer accounts for it. The defensive branch needs no debit of its own, for the same reason: the first removal already made it — a comment in that branch now records the invariant.
• Apply the identical move and comment in PersistentDispatcherMultipleConsumersClassic.
• No change to the available-permits accounting on either dispatcher: that half of the removal path belongs to [fix][broker] Fix persistent throughput degradation caused by permit loss during frequent reconnects on Shared subscriptions #26289.
• PersistentStickyKeyDispatcherMultipleConsumers (Key_Shared) inherits the fix through super.removeConsumer.

Verifying this change

This change added tests and can be verified as follows:

• SharedSubscriptionUnackedMessagesAccountingTest#testRemovingSameConsumerTwiceDebitsUnackedMessagesOnce — carried over verbatim from [fix][broker] Debit only credited permits when removing a Shared consumer #26417: leaves a consumer holding ten un-acknowledged deliveries, removes it twice, and requires the subscription counter to end at zero. Fails on master (expected [0] but found [-10]), passes with this change.
• SharedSubscriptionUnackedMessagesAccountingTest#testRemovingSameConsumerTwiceDebitsUnackedMessagesOnceOnClassicDispatcher — the same probe against the classic dispatcher, flipped in for the duration of the test through the dynamic flag and restored afterwards. Fails on master with the same -10, passes with this change.

Both tests were run red-first against the unmodified production code, then green with the fix applied.

branch-4.0 and branch-4.2 carry the same unguarded debit in both dispatchers, so this should backport cleanly; the tests ride along for that purpose.

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

If the box was checked, please highlight the changes

• 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

[lhotari]

Thanks for splitting this out of #26417 — the carve-out was the right call, and the write-up made the review easy: the motivation, the invariant and the red-on-master evidence are all stated precisely. Welcome, and nice first patch.

I verified the core claim independently against the code at fbf8ca7:

• Consumer.unackedMessages is only mutated by addAndGetUnAckedMsgs and clearUnAckedMsgs, and neither runs on the dispatcher removal path, so after the first removeConsumer the departing consumer still reports its full count. clearComponentsAfterRemovedAllConsumers() resets totalAvailablePermits but not totalUnackedMessages. On master the second removal therefore really does land the subscription at -10, exactly as your test asserts (10 → 0, then 0 → -10).
• The double removal is reachable outside a test: Consumer.close(boolean) calls subscription.removeConsumer(this, …) unconditionally, and #22270 exists precisely because the consumerSet/consumerList mismatch shows up in the field — the branch even logs at ERROR.
• Moving the debit after consumerSet.removeAll opens no re-entrancy window. Every unblock path reachable from addUnAckedMessages goes through readMoreEntriesAsync(), including BrokerService.unblockDispatchersOnUnAckMessages, so nothing can dispatch while consumerList still holds the departing consumer. Consumer.sendMessages runs under the same dispatcher monitor as removal.
• The scope is right: PersistentStickyKeyDispatcherMultipleConsumers and its classic counterpart inherit the fix through super.removeConsumer, and NonPersistentDispatcherMultipleConsumers does not track un-acked messages at all.
• No conflict with #26289: it edits the same method at old lines 258-266 and 300-323, this PR at old lines 239-247 and 274-279. Disjoint hunks, different aggregates — they compose.

One candidate defect I chased down and ruled out: that consumerSet.removeAll(consumer) could return more than 1 for a consumer registered twice (the #22283 warning path), sending the removal into the un-debited else branch. consumerSet is a com.carrotsearch.hppc.ObjectHashSet (AbstractDispatcherMultipleConsumers:35) and ObjectHashSet.removeAll(KType) is literally return remove(key) ? 1 : 0, so it can only return 0 or 1. In the duplicate-registration case consumerList holds two entries and consumerSet one: the first removal returns 1 and debits, the second returns 0 and does not — a single debit for a single Consumer, which is what you want. Good.

The production change looks right to me. What I would still like addressed is on the test side, plus one accuracy point about the new comment — details inline.

One caveat on my side, so you know exactly how far my check goes: I did not run the new tests locally. The red-on-master claim above is derived from reading the code rather than from an observed failing run, and the PR's own CI is green on fbf8ca7.

[lhotari]

The review feedback is addressed. Both dispatcher variants now use dedicated brokers and verify subscription and broker-wide unacked-message accounting after each removal. The removal comments and test assertions have also been updated.

Validation: both dispatcher tests and quickCheck passed. Both variants fail with the original unconditional debit, and both detect the broker-wide double debit when only the subscription counter is reset.

[FlorentinDUBOIS]

Thanks a lot @lhotari!