[improve][broker]Optimize InMemoryDelayedDeliveryTracker by maintaining state #23918
Conversation
yunmaoQu
changed the title
| @Override | ||
| public CompletableFuture<Void> clear() { | ||
| this.delayedMessageMap.clear(); | ||
| long cutoffTime = getCutoffTime(); |
There was a problem hiding this comment.
@dao-jun The decision to clear only expired indices in the clear() method of the InMemoryDelayedDeliveryTracker is aimed at optimizing performance and maintaining the logical state of the message delivery system.
By focusing on expired messages, we reduce the overhead associated with clearing and re-adding valid messages, which enhances performance, especially in scenarios with a high volume of delayed messages. This approach also allows us to retain the state of valid messages, enabling more efficient message delivery without needing to re-read them from storage.
There was a problem hiding this comment.
I doubt that it will introduce memory leak problem. For example, we need to clear all delayed messages at method:
org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers#clearDelayedMessages
IMO, the payoff of re-adding valid messages for once is acceptable, while the risk of such change is great.
| @@ -218,9 +218,26 @@ public NavigableSet<Position> getScheduledMessages(int maxMessages) { | |||
| return positions; | |||
| } | |||
|
|
|||
| public boolean shouldSkipMessage(long ledgerId, long entryId) { | |||
There was a problem hiding this comment.
it better to keep the method name with BucketDelayedDeliveryTracker, change shouldSkipMessage to contains.
|
Hi @dao-jun , I'm in need of your review on this PR to move forward with the feature. Could you please check it out as soon as you can? Thanks a lot! |
There was a problem hiding this comment.
Could you help to add unit test to cover the logic changed?
We need to ensure that such kind of skipping reading delayed messages won't impact the normal read when we want to dispatch them.
And i wonder that how much improvement this pr can offer? Do unnecessary reads occur frequently or not? In what cases the unnecessary reads will occur? If the answer is positive, i am glad to accept this improvement and adopt it in my production.
@yunmaoQu @lhotari
| @Override | ||
| public CompletableFuture<Void> clear() { | ||
| this.delayedMessageMap.clear(); | ||
| long cutoffTime = getCutoffTime(); |
There was a problem hiding this comment.
I doubt that it will introduce memory leak problem. For example, we need to clear all delayed messages at method:
org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers#clearDelayedMessages
IMO, the payoff of re-adding valid messages for once is acceptable, while the risk of such change is great.
| if (deliveryTracker instanceof BucketDelayedDeliveryTracker) { | ||
| if (deliveryTracker instanceof InMemoryDelayedDeliveryTracker) { | ||
| skipCondition = position -> ((InMemoryDelayedDeliveryTracker) deliveryTracker) | ||
| .shouldSkipMessage(position.getLedgerId(), position.getEntryId()); | ||
| } else if (deliveryTracker instanceof BucketDelayedDeliveryTracker) { | ||
| skipCondition = position -> ((BucketDelayedDeliveryTracker) deliveryTracker) | ||
| .containsMessage(position.getLedgerId(), position.getEntryId()); |
There was a problem hiding this comment.
change shouldSkipMessage to contains to avoid duplicate code.
Fixes #23912
Main Issue: #23912
PIP: #xyz
Motivation
In the current delayed message delivery, there's an opportunity to reduce unnecessary reads to storage.
Modifications
It would be useful to keep state also in the InMemoryDelayedDeliveryTracker and skip reading delayed messages when the information is already available for the delivery time of a specific entry.
Verifying this change
This change is already covered by existing tests.
Does this pull request potentially affect one of the following parts:
Documentation
docdoc-requireddoc-not-neededdoc-completeMatching PR in forked repository
PR in forked repository:
yunmaoQu#2