[improve][broker][PIP-379] Add observability stats for "draining hashes" #23429
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…ith array format The String format is very inefficient. It's better to replace it for PIP-379 This is needed for tests
lhotari
requested review from
merlimat,
equanz,
codelipenghui and
poorbarcode
github-actions
bot
added
the
doc-required
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #23429 +/- ##
============================================
+ Coverage 73.57% 74.33% +0.76%
- Complexity 32624 34398 +1774
============================================
Files 1877 1950 +73
Lines 139502 146980 +7478
Branches 15299 16184 +885
============================================
+ Hits 102638 109261 +6623
- Misses 28908 29298 +390
- Partials 7956 8421 +465
Flags with carried forward coverage won't be shown. Click here to find out more.
|
Technoboy-
reviewed
Oct 10, 2024
...broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
Show resolved
Hide resolved
2 tasks
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Motivation
PIP-379: Key_Shared Draining Hashes for Improved Message Ordering was implemented in #23352.
One of the major benefits of PIP-379 is the easy-to-understand model of when a hash is blocked.
When a new consumer is added, hash range assignments move from existing consumers to the new consumer. (In some cases, hash range assignments can move between existing consumers after a consumer is added or removed.)
The PIP-379 implementation ensures that no new messages for the hash ranges that were moved can be delivered until all unacknowledged messages for a specific hash are cleared with acknowledgements or when the consumer disconnects.
This applies to the AUTO_SPLIT ordered mode of the Key_Shared subscription type.
There's a concept of "draining hashes" in PIP-379 which is now reflected in the consumer stats. This is an intentionally exposed internal detail since the user must have the information available for understanding why messages don't get delivered.
Since there's no mapping between external and internal concepts, the abstraction isn't leaky. The user doesn't need to know about the internal details of how the draining hashes are implemented, but they need to know that the consumer is blocked on unacknowledged messages for a specific hash range. This is all relevant information and doesn't contain unnecessary implementation details.
This PR contains the "consumer stats" changes that provide the information in a clear way.
Modifications
Added consumer-level stats:
drainingHashesCount- the current number of hashes in the draining state for this consumerdrainingHashesClearedTotal- the total number of hashes cleared from the draining state since the consumer connecteddrainingHashesUnackedMessages- the total number of unacknowledged messages for all draining hashes for this consumerdrainingHashes- draining hashes information for this consumerhash- the sticky key hash which is drainingunackMsgs- the number of unacknowledged messages for this hashblockedAttempts- the number of times the hash has blocked an attempted delivery of a messageIn addition:
keyHashRangeArrays- the consumer's hash range assignments in a list of lists where each item contains the start and end as elements.[ [ 2960, 5968 ], [ 22258, 43033 ], [ 49261, 54464 ], [ 55155, 61273 ] ]It was necessary to add this field with a new name
keyHashRangeArrayssince there's already an existingkeyHashRangefield. Changing that isn't possible since it would break compatibility. A newer admin client couldn't read stats from an older broker and vice-versa.The previous
keyHashRangeis now deprecated. The field format was different.Example of both fields where the difference is visible:
{ "keyHashRangeArrays" : [ [ 2960, 5968 ], [ 22258, 43033 ], [ 49261, 54464 ], [ 55155, 61273 ] ], "keyHashRanges" : [ "[2960, 5968]", "[22258, 43033]", "[49261, 54464]", "[55155, 61273]" ], }The field
keyHashRangescontains the information as a list of string values, which isn't very usable for most use cases since it would need to be parsed before it can be used.The stats will continue to contain
keyHashRangeandreadPositionWhenJoiningwhen the "classic" (3.3.x) implementation of Key_Shared is used by configuringsubscriptionKeySharedUseClassicPersistentImplementation=true("classic" support was added in #23424).In the default configuration, the fields are removed from the topic stats output, but the client continues to support the fields for backward and forward compatibility.
Example of consumer stats for a subscription
{ "consumers" : [ { "msgRateOut" : 0.0, "msgThroughputOut" : 0.0, "bytesOutCounter" : 1560, "msgOutCounter" : 30, "msgRateRedeliver" : 0.0, "messageAckRate" : 0.0, "chunkedMessageRate" : 0.0, "consumerName" : "c1", "availablePermits" : 70, "unackedMessages" : 30, "avgMessagesPerEntry" : 1, "blockedConsumerOnUnackedMsgs" : false, "drainingHashesCount" : 5, "drainingHashesClearedTotal" : 0, "drainingHashesUnackedMessages" : 10, "drainingHashes" : [ { "hash" : 2862, "unackMsgs" : 2, "blockedAttempts" : 5 }, { "hash" : 11707, "unackMsgs" : 2, "blockedAttempts" : 9 }, { "hash" : 15786, "unackMsgs" : 2, "blockedAttempts" : 6 }, { "hash" : 43539, "unackMsgs" : 2, "blockedAttempts" : 6 }, { "hash" : 45436, "unackMsgs" : 2, "blockedAttempts" : 9 } ], "address" : "/127.0.0.1:55829", "connectedSince" : "2024-10-10T05:39:39.077284+03:00", "clientVersion" : "Pulsar-Java-v4.0.0-SNAPSHOT", "lastAckedTimestamp" : 0, "lastConsumedTimestamp" : 1728527979411, "lastConsumedFlowTimestamp" : 1728527979106, "keyHashRangeArrays" : [ [ 2960, 5968 ], [ 22258, 43033 ], [ 49261, 54464 ], [ 55155, 61273 ] ], "metadata" : { }, "lastAckedTime" : "1970-01-01T02:00:00+02:00", "lastConsumedTime" : "2024-10-10T05:39:39.411+03:00" }, { "msgRateOut" : 0.0, "msgThroughputOut" : 0.0, "bytesOutCounter" : 0, "msgOutCounter" : 0, "msgRateRedeliver" : 0.0, "messageAckRate" : 0.0, "chunkedMessageRate" : 0.0, "consumerName" : "c2", "availablePermits" : 1000, "unackedMessages" : 0, "avgMessagesPerEntry" : 0, "blockedConsumerOnUnackedMsgs" : false, "drainingHashesCount" : 0, "drainingHashesClearedTotal" : 0, "drainingHashesUnackedMessages" : 0, "drainingHashes" : [ ], "address" : "/127.0.0.1:55829", "connectedSince" : "2024-10-10T05:39:39.294216+03:00", "clientVersion" : "Pulsar-Java-v4.0.0-SNAPSHOT", "lastAckedTimestamp" : 0, "lastConsumedTimestamp" : 0, "lastConsumedFlowTimestamp" : 1728527979297, "keyHashRangeArrays" : [ [ 1, 2959 ], [ 5969, 22257 ], [ 43034, 49260 ], [ 54465, 55154 ], [ 61274, 65535 ] ], "metadata" : { }, "lastAckedTime" : "1970-01-01T02:00:00+02:00", "lastConsumedTime" : "1970-01-01T02:00:00+02:00" } ] }Relevant information for consumer c1:
{ "drainingHashesCount" : 5, "drainingHashesClearedTotal" : 0, "drainingHashesUnackedMessages" : 10, "drainingHashes" : [ { "hash" : 2862, "unackMsgs" : 2, "blockedAttempts" : 5 }, { "hash" : 11707, "unackMsgs" : 2, "blockedAttempts" : 9 }, { "hash" : 15786, "unackMsgs" : 2, "blockedAttempts" : 6 }, { "hash" : 43539, "unackMsgs" : 2, "blockedAttempts" : 6 }, { "hash" : 45436, "unackMsgs" : 2, "blockedAttempts" : 9 } ], }Relevant information in this case about consumer c2:
{ "keyHashRangeArrays" : [ [ 1, 2959 ], [ 5969, 22257 ], [ 43034, 49260 ], [ 54465, 55154 ], [ 61274, 65535 ] ], }The PIP-379 implementation will only block hashes that are necessary. For each hash, there's a way to get detailed information to find out why the delivery is blocked.
The major difference from the previous
readPositionWhenJoiningsolution is that it's possible to automate and build CLI and web user interface tools to assist a user, making it very easy to troubleshoot issues when message delivery is blocked by unacknowledged messages in Key_Shared subscriptions.Client-side tooling could already use the information provided in this PR to determine which consumer is blocked by a hash in the case that there would be multiple consumers.
In the above example, the hash
2862is contained in the hash range[1, 2959], which means 2 unacknowledged messages for that hash are preventing further messages with hash2862from being delivered to consumerc2.The
blockedAttemptsfield contains a counter that increments each time the dispatcher skips delivery to a consumer due to this hash. Using this information alone, it's very convenient to observe Key_Shared AUTO_SPLIT subscriptions and find out the causes.A future improvement will be to add a REST API for finding out the unacknowledged message ID information of the unacknowledged message for a hash. Using this information, it's possible to find out the details of the message that is blocking a particular hash.
Documentation
docdoc-requireddoc-not-neededdoc-complete