[fix][broker] Prevent stale consumer removal from removing replacements

[Denovo1998]

Follow-up to #26422.
Related to #26289.

Motivation

A delayed removal of an old consumer can match a replacement consumer through Consumer.equals(), even though they are different instances. This can remove the replacement from dispatcher membership and debit the old consumer's unacknowledged messages again. For Key_Shared subscriptions, the stale removal can also remove the replacement from the selector and prevent further delivery.

Modifications

• Require the exact registered consumer instance before removing dispatcher membership or debiting its unacknowledged messages.
• Remove consumer list entries by identity while maintaining priority counts, including defensive cleanup and failed-add cleanup.
• Protect replacement consumers from stale AUTO_SPLIT removals.
• Preserve cleanup for STICKY additions that complete after dispatcher removal. Remove selector ranges and classic recently-joined records by identity.
• Apply the changes to both classic and newer persistent dispatchers.

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

• Added integration tests for end-to-end deployment with large payloads (10MB)
• Extended integration test for recovery after broker failure

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 protecting replacement consumers across both dispatcher implementations. The identity checks look sound for completed additions, but the new Key_Shared early return drops cleanup for a STICKY addition that completes after removal. Please preserve that cleanup and cover the delayed-add case.

[lhotari]

[BUG] Keep cleanup for STICKY additions that finish after removal

An unregistered consumer can still have selector state: HashRangeExclusiveStickyKeyConsumerSelector.addConsumer waits on a conflicting consumer's liveness check before installing its range (HashRangeExclusiveStickyKeyConsumerSelector.java:59-79).

Consider a pending STICKY subscribe on another connection that times out while this check is outstanding. A concurrent cursor reset removes both consumers from the dispatcher. If the liveness check then completes, the selector can install the removed consumer's range. The original subscribe future is already exceptional, so ServerCnx calls consumer.close() again (ServerCnx.java:2152-2155). This new early return skips the selector cleanup that previously ran on that second close, leaving an orphan range that blocks later same-range subscriptions. The classic guard at PersistentStickyKeyDispatcherMultipleConsumersClassic.java:237-240 has the same issue.

Please invalidate pending additions when removing a consumer, or preserve exact-instance selector cleanup for this case while protecting equal replacements. A regression test can hold the STICKY liveness future, remove both consumers, complete the add, then repeat the close and verify that a new same-range subscribe succeeds.

[Denovo1998]

@lhotari Indeed, there is this issue. I just dealt with it, could you please review it again and take a look?