[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