[fix][broker] Key_Shared subscription doesn't always deliver messages from the replay queue after a consumer disconnects and leaves a backlog

[nborisov]

Fixes apache#23845

Motivation

There is a not covered scenario for draining hashes and key shared subscriptions at PersistentStickyKeyDispatcherMultipleConsumers. The detailed scenario described at apache#23845 (comment)
As a result draining hashes could contain entries with consumer which was stopped. This leads consuming to get stuck.

Modifications

Decrease draining hash entry refCount in case its hash range returned to the initial owner which holds the entry in pending acks.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Added integration test org.apache.pulsar.client.api.KeySharedSubscriptionTest#testMessageDeliveredFromDrainingHashes to verify the scenario
• Existing unit tests were modified to check changes applied in PR: org.apache.pulsar.broker.service.ConsistentHashingStickyKeyConsumerSelectorTest#testShouldNotSwapExistingConsumers, org.apache.pulsar.broker.service.ConsumerHashAssignmentsSnapshotTest#testResolveConsumerRemovedHashRanges_NoChanges, org.apache.pulsar.broker.service.ConsumerHashAssignmentsSnapshotTest#testResolveConsumerUpdatedHashRanges_RangeAdded, org.apache.pulsar.broker.service.ConsumerHashAssignmentsSnapshotTest#testResolveConsumerRemovedHashRanges_RangeUpdated, org.apache.pulsar.broker.service.ConsumerHashAssignmentsSnapshotTest#testResolveConsumerUpdatedHashRanges_OverlappingRanges

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

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: #4