diff --git a/pip/pip-393.md b/pip/pip-393.md new file mode 100644 index 0000000000000..646c2beb5fe40 --- /dev/null +++ b/pip/pip-393.md @@ -0,0 +1,226 @@ + +# PIP-393: Improve performance of Negative Acknowledgement + +# Background knowledge + +Negative Acknowledgement is a feature in Pulsar that allows consumers to trigger the redelivery +of a message after some time when they fail to process it. When user calls `negativeAcknowledge` method, +`NegativeAcksTracker` in `ConsumerImpl` will add an entry into the map `NegativeAcksTracker.nackedMessages`, +mapping the message ID to the redelivery time. When the redelivery time comes, `NegativeAcksTracker` will +send a redelivery request to the broker to redeliver the message. + +# Motivation + +There are many issues with the current implementation of Negative Acknowledgement in Pulsar: +- the memory occupation is high. +- the code execution efficiency is low. +- the redelivery time is not accurate. +- multiple negative ack for messages in the same entry(batch) will interfere with each other. +All of these problem is severe and need to be solved. + +## Memory occupation is high +After the improvement of https://github.com/apache/pulsar/pull/23582, we have reduce half more memory occupation +of `NegativeAcksTracker` by replacing `HashMap` with `ConcurrentLongLongPairHashMap`. With 1 million entry, the memory +occupation decrease from 178MB to 64MB. With 10 million entry, the memory occupation decrease from 1132MB to 512MB. +The average memory occupation of each entry decrease from 1132MB/10000000=118byte to 512MB/10000000=53byte. + +But it is not enough. Assuming that we negative ack message 10k/s, assigning 1h redelivery delay for each message, +the memory occupation of `NegativeAcksTracker` will be `3600*10000*53/1024/1024/1024=1.77GB`, if the delay is 5h, +the required memory is `3600*10000*53/1024/1024/1024*5=8.88GB`, which increase too fast. + +## Code execution efficiency is low +Currently, each time the timer task is triggered, it will iterate all the entries in `NegativeAcksTracker.nackedMessages`, +which is unnecessary. We can sort entries by timestamp and only iterate the entries that need to be redelivered. + +## Redelivery time is not accurate +Currently, the redelivery check time is controlled by the `timerIntervalNanos`, which is 1/3 of the `negativeAckRedeliveryDelay`. +That means, if the `negativeAckRedeliveryDelay` is 1h, check task will be started every 20min, the deviation of the redelivery +time is 20min, which is unacceptable. + +## Multiple negative ack for messages in the same entry(batch) will interfere with each other +Currently, `NegativeAcksTracker#nackedMessages` map `(ledgerId, entryId)` to `timestamp`, which means multiple nacks from messages +in the same batch share single one timestamp. +If we let msg1 redelivered 10s later, then let msg2 redelivered 20s later, these two messages are delivered 20s later together. +msg1 will not be redelivered 10s later as the timestamp recorded in `NegativeAcksTracker#nackedMessages` is overrode by the second +nack call. + + +# Goals + +Refactor the `NegativeAcksTracker` to solve the above problems. + +To avoid interation of all entries in `NegativeAcksTracker.nackedMessages`, we use a sorted map to store the entries. +To reduce memory occupation, we use util class provided by fastutil(https://fastutil.di.unimi.it/docs/), and design +a new algorithm to store the entries, reduce the memory occupation to even 1% less than the current implementation. +(the actual effect rely on the configuration and the throughput). + +# Detailed Design + +## Design & Implementation Details + +### New Data Structure +Use following data structure to store the entries: +```java +Long2ObjectSortedMap> nackedMessages = new Long2ObjectAVLTreeMap<>(); +``` +mapping `timestamp -> ledgerId -> entryId`. +We need to sort timestamp in ascending order, so we use a sorted map to map timestamp to `ledgerId -> entryId` map. +As there will many entries in the map, we use `Long2ObjectAVLTreeMap` instead of `Long2ObjectRBTreeMap`. +As for the inner map, we use `Long2ObjectMap` to map `ledgerId` to `entryId` because we don't need to keep the order of `ledgerId`. +`Long2ObjectOpenHashMap` will be satisfied. +All entry id for the same ledger id will be stored in a bit set, as we only care about the existence of the entry id. + + +### TimeStamp Bucket +Timestamp in ms is used as the key of the map. As most of the use cases don't require that the precision of the delay time is 1ms, +we can make the timestamp bucketed, that is, we can trim the lower bit of the timestamp to map the timestamp to a bucket. +For example, if we trim the lower 1 bit of the timestamp, the timestamp 0b1000 and 0b1001 will be mapped to the same bucket 0b1000. +Then all messages in the same bucket will be redelivered at the same time. +If user can accept 1024ms deviation of the redelivery time, we can trim the lower 10 bits of the timestamp, which can group a lot +entries into the same bucket and reduce the memory occupation. + +following code snippet will be helpful to understand the design: +```java + static long trimLowerBit(long timestamp, int bits) { + return timestamp & (-1L << bits); + } +``` + +```java +Long2ObjectSortedMap> map = new Long2ObjectAVLTreeMap<>(); +Long2ObjectMap ledgerMap = new Long2ObjectOpenHashMap<>(); +LongSet entrySet = new LongOpenHashSet(); +entrySet.add(entryId); +ledgerMap.put(ledgerId, entrySet); +map.put(timestamp, ledgerMap); +``` + +### Configuration + +Add a new configuration `negativeAckPrecisionBitCnt` to control the precision of the redelivery time. +``` +@ApiModelProperty( + name = "negativeAckPrecisionBitCnt", + value = "The redelivery time precision bit count. The lower bits of the redelivery time will be\n" + + "trimmed to reduce the memory occupation. The default value is 8, which means the redelivery time\n" + + "will be bucketed by 256ms. In worst cases, the redelivery time will be 512ms earlier(no later)\n" + + "than the expected time. If the value is 0, the redelivery time will be accurate to ms.". + ) + private long negativeAckPrecisionBitCnt = 8; +``` +The higher the value, the more entries will be grouped into the same bucket, the less memory occupation, the less accurate the redelivery time. +Default value is 8, which means the redelivery time will be bucketed by 256ms. In worst cases, the redelivery time will be 512ms earlier(no later) +than the expected time. + + +## Space complexity analysis +### Space complexity of `ConcurrentLongLongPairHashMap` +Before analyzing the new data structure, we need to know how much space it take before this pip. + +We need to store 4 long field for `(ledgerId, entryId, partitionIndex, timestamp)` for each entry, which takes `4*8=32byte`. +As `ConcurrentLongLongPairHashMap` use open hash addressing and linear probe to handle hash conflict, there are some +redundant spaces to avoid high conflict rate. There are two configurations that control how much redundant space to reserver: +`fill factor` and `idle factor`. When the space utility rate soar high to `fill factor`, the size of backing array will +be double, when the space utility rate reduce to `idle factor`, the size of backing array will reduce by half. + +The default value of `fill factor` is 0.66, `idle factor` is 0.15, which means the min space occupation of +`ConcurrentLongLongPairHashMap` is `32/0.66N byte = 48N byte`, the max space occupation is `32/0.15N byte=213N byte`, +where N is the number of entries. + +In the experiment showed in the PR, there are 1 million entries in the map, taking up `32*1000000/1024/1024byte=30MB`, +the space utility rate is 30/64=0.46, in the range of `[0.15, 0.66]`. + + +### Space complexity of the new data structure +The space used by new data structure is related to several factors: `message rate`, `the time deviation user accepted`, +`the max entries written in one ledger`. +- Pulsar conf `managedLedgerMaxEntriesPerLedger=50000` determine the max entries can be written into one ledger, +we use the default value to analyze. +- `the time deviation user accepted`: when user accept 1024ms delivery time deviation, we can trim the lower 10 bit +of the timestamp in ms, which can bucket 1024 timestamp. + +Following we will analyze the space used by one bucket, and calculate the average space used by one entry. + +Assuming that the message rate is `x msg/ms`, and we trim `y bit` of the timestamp, one bucket will contains `2**x` ms, and +`M=2**x*y` msgs. +- For one single bucket, we only need to store one timestamp, which takes `8byte`. +- Then, we need to store the ledgerId, when M is greater than 5w(`managedLedgerMaxEntriesPerLedger`), the ledger will switch. +There are `L=ceil(M/50000)` ledgers, which take `8*L` byte. +- Further, we analyze how much space the entry id takes. As there are `L=ceil(M/50000)` ledgers, there will be `L` bitmap to store, +which take `L*size(bitmap)`. The total space consumed by new data structure is `8byte + 8L byte + L*size(bitmap)`. + +As the `size(bitmap)` is far more greater than `8byte`, we can ignore the first two items. Then we get the formular of space +consumed **one bucket**: `D=L*size(bitmap)=ceil(M/50000)*size(bitmap)`. + +Entry id is stored in a `Roaring64Bitmap`, for simplicity we can replace it with `RoaringBitmap`, as the max entry id is 49999, +which is smaller than `4294967296 (2 * Integer.MAX_VALUE)`(the max value can be stored in `RoaringBitmap`). The space consume +by `RoaringBitmap` depends on how many elements it contains, when the size of bitmap < 4096, the space is `4N byte`, +when the size of bitmap > 4096, the consumed space is a fixed value `8KB`. + +Then we get the final result: +- when M>50000, `D = ceil(M/50000)*size(bitmap) ~= M/50000 * 8KB = M/50000 * 8 * 1024 byte = 0.163M byte`, +each entry takes `0.163byte` by average. +- when 409650000, `0.163N byte`. + - when 4096 +* Mailing List discussion thread: https://lists.apache.org/thread/yojl7ylk7cyjxktq3cn8849hvmyv0fg8 +* Mailing List voting thread: https://lists.apache.org/thread/hyc1r2s9chowdhck53lq07tznopt50dy