[pulsar-broker] Optimize message replay for large backlog consumer #3732
Conversation
| } | ||
|
|
||
| @Override | ||
| public <T> Set<T> items(int numberOfItems, BiFunction<Long, Long, T> longPairConverter) { |
There was a problem hiding this comment.
If we use some kind a specialized interface we can avoid 2 long -> Long conversions:
interface LongPairFunction<T> {
T apply(long a, long b);
}There was a problem hiding this comment.
yes, I wanted to create it custom interface to reduce number of obj creation but then I missed it. Good catch. I will fix it.
|
|
||
| @Override | ||
| public <T> Set<T> items(int numberOfItems, BiFunction<Long, Long, T> longPairConverter) { | ||
| Set<T> items = new HashSet<>(); |
There was a problem hiding this comment.
Is it necessary to return a set? typically the iteration methods are used to go through it inline without intermediary collection.
There was a problem hiding this comment.
yes, this method is used by PersistentDispatcherMultipleConsumers to prepare the sub-set of replay-messages and then it passes this set to cursor.asyncReplayEntries to read the entries. So, this method is not used for inline execution.
| @@ -288,8 +287,8 @@ public void readMoreEntries() { | |||
| return; | |||
| } | |||
|
|
|||
| Set<PositionImpl> messagesToReplayNow = messagesToReplay.items(messagesToRead).stream() | |||
| .map(pair -> new PositionImpl(pair.first, pair.second)).collect(toSet()); | |||
| Set<PositionImpl> messagesToReplayNow = messagesToReplay.items(messagesToRead, | |||
There was a problem hiding this comment.
Instead of having the messagesToReplay to be sorted, wouldn't be easier to just make the messagesToReplayNow to be a SortedSet ?
In this case, we don't care to have messagesToReplay to be always sorted, just when there is a redelivery is when we care.
There was a problem hiding this comment.
wouldn't be easier to just make the messagesToReplayNow to be a SortedSet ?
No because messagesToReplayNow is a very small sub-set (100 msgs) of messagesToReplay (> 1M msgs) and the main issue is messagesToReplay can have more than million random messages and we want to read messages from the same ledger. making ONLY messagesToReplayNow sorted set will not help because there is a possibility that 100 read messages from messagesToReplay might be random from different ledgers and it will still perform random read across multiple ledgers.
|
rerun java8 tests |
|
rerun java8 tests |
|
rerun java8 tests |
1 similar comment
|
rerun java8 tests |
add tests fix items arg with logPairFunction interface
|
rerun integration tests |
Motivation
It address #3731.
Modification
ConcurrentSortedLongPairSetthat helps broker to avoid random read across multiple managed-ledgers. It still usesConcurrentLongPairSetto avoid object allocation for message-ids.<T> Set<T> items(int numberOfItems, BiFunction<Long, Long, T> longPairConverter)method inLongPairSetto avoid creating temporary LongPair objects.Note
I have done performance testing on
GrowablePriorityLongPairQueuewhich I had introduced sometime back but its insert/remove is super slow and CPU intensive so, we can't use it in this scenario.