[improve][pip] PIP-393: Improve performance of Negative Acknowledgeme…

…nt (apache#23601)

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<Long2ObjectMap<Roaring64Bitmap>> 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<Long2ObjectMap<Roaring64Bitmap>> map = new Long2ObjectAVLTreeMap<>();
+Long2ObjectMap<LongSet> 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 4096<M<50000, `D = ceil(M/50000)*size(bitmap) = 1 * 8KB = 8KB`, each entry takes `8*1024/M=8192/M byte` by average.
+- when M<4096, `D = ceil(M/50000)*size(bitmap) = 1 * 4M byte = 4M byte`, each entry take `4 byte` by average.
+
+### Conclusion
+Assuming N is the number of entries, M is the number of messages in one bucket.
+- `ConcurrentLongLongPairHashMap`: `48N` byte in best case, `213N byte` in worst case.
+- New data structure:
+    - when M>50000, `0.163N byte`.
+    - when 4096<M<50000, `8192/M * N byte` .
+    - when M<4096, `4N byte`.
+
+Some experiment results are showed in the PR, we can fine tune the configuration to get the best performance.
+
+## Effect
+
+### Memory occupation is high
+With such kind of design, we can reduce the memory occupation of `NegativeAcksTracker` to 1% less than the current implementation.
+
+### Code execution efficiency is low
+With the new design, we can avoid the iteration of all entries in `NegativeAcksTracker.nackedMessages`, and only iterate the entries
+that need to be redelivered.
+
+### Redelivery time is not accurate
+With the new design, we avoid the fixed interval of the redelivery check time. We can control the precision of the redelivery time 
+by trimming the lower bits of the timestamp. 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
+
+### Multiple negative ack for messages in the same entry(batch) will interfere with each other
+With the new design, if we let msg1 redelivered 10s later, then let msg2 redelivered 20s later, these two nacks will not interfere
+with each other, as they are stored in different buckets.
+
+
+## High-Level Design
+As this pip introduce new dependency `fastutil` into client, which is very large(23MB), while few classes are used, we need to
+reduce the size of the dependency. 
+
+Though there is alternative dependency `fastutil-core`, which is smaller(6MB), but it is also
+relatively large and using `fastutil-core` will introduce another problem on the broker side since there's already `fastutil` jar 
+which also includes `fastutil-core` jar classes.
+
+The optimal solution would be to include only the classes from fastutil into the shaded pulsar-client and pulsar-client-all 
+which are really used and needed. This could be achieved in many ways. One possible solution is to introduce an intermediate
+module for shaded pulsar-client and pulsar-client-all that isn't published to maven central at all. 
+It would be used to minimize and include only the classes from fastutil which are required by pulsar-client shading.
+
+
+
+# Backward & Forward Compatibility
+
+## Upgrade
+
+User can upgrade to the new version without any compatibility issue.
+
+## Downgrade / Rollback
+
+User can downgrade to the old version without any compatibility issue.
+
+# Links
+
+<!--
+Updated afterwards
+-->
+* Mailing List discussion thread: https://lists.apache.org/thread/yojl7ylk7cyjxktq3cn8849hvmyv0fg8
+* Mailing List voting thread: https://lists.apache.org/thread/hyc1r2s9chowdhck53lq07tznopt50dy