[improve][pip] PIP-426: Enhanced Consumer Throttling and Unacknowledged Message Tracking for Exclusive and Failover Subscriptions

[berg223]

Fixes #24159

Main Issue: #24159

Motivation

1. I have tried to fix the issue by [fix][broker] fix unacked message count is zero when using exclusive subscription #24376. However, that PR doesn't support features about cumulative ack, batching mode, and transaction. So I want to improve it further.
2. I have found another issue that flowcontrol of exclusive or failover consumer is not work. Since the issues are highly correlated, I want to fix them at the same time.

Modifications

1. Remove the Subscription.isIndividualAckMode() restriction that limits pendingAcks usage to shared subscriptions.
2. Extend PendingAckHandleImpl for exclusive/failover consumers.
3. Enhance flow control in PersistentDispatcherSingleActiveConsumer

Verifying this change

• Make sure that the change passes the CI checks.

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:

[lhotari]

Yes, I agree with this.

After having "pending acks" also for failover subscriptions, it will be easier to come up with a proper solution for #15189 . When a failover subscription changes the active consumer, it causes duplicate processing which is problematic and surprising for many applications. Failover subscription active consumer changing should be potentially handled by first "draining" the previous consumer before sending messages to the new active consumer. (The draining approach is used for Key_Shared subscriptions since Pulsar 4.0 / PIP-379)

[berg223]

Yes. It's also benefit to solve similiar issues to #15189. The draining approach could also be implemented . Your comments always open my mind and point me in the right direction !

[codelipenghui]

Indeed, each message in a Pulsar topic can have a persistent, monotonically increasing local index, which functions much like a Kafka offset. This feature was officially introduced by Pull Request #9039, which implemented PIP-70 ("Introduce lightweight broker entry metadata").

With this local index, the number of unacknowledged messages for a consumer can be calculated with a simple subtraction:

unacked_messages = last_delivered_index - mark_delete_position

Given its reliability, considering its proven value, enabling this index by default in a future Apache Pulsar release would be reasonable.

[berg223]

Thanks for more background knowledge! It's really helpful to solve origin issue #24159 . Since there is no need to add flow permits to exlusive or failover subscription (details discussion see: https://lists.apache.org/thread/glvd8lrvyll9mdtp62d34x1k38swrls1), there is no motivation except fix the origin issue. IMO, we can close this pip to keep the pip from too much useless information.