Denovo1998 commented on code in PR #25076:
URL: https://github.com/apache/pulsar/pull/25076#discussion_r2623376743
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java:
##########
@@ -126,10 +126,13 @@ public boolean addMessage(long ledgerId, long entryId,
long deliverAt) {
}
long timestamp = trimLowerBit(deliverAt, timestampPrecisionBitCnt);
- delayedMessageMap.computeIfAbsent(timestamp, k -> new
Long2ObjectRBTreeMap<>())
- .computeIfAbsent(ledgerId, k -> new Roaring64Bitmap())
- .add(entryId);
- delayedMessagesCount.incrementAndGet();
+ Roaring64Bitmap roaring64Bitmap = delayedMessageMap
+ .computeIfAbsent(timestamp, k -> new Long2ObjectRBTreeMap<>())
+ .computeIfAbsent(ledgerId, k -> new Roaring64Bitmap());
+ if (!roaring64Bitmap.contains(entryId)) {
+ roaring64Bitmap.add(entryId);
Review Comment:
Threading question: addMessage() and getScheduledMessages() are invoked
under synchronized (this) in the dispatcher (e.g.
PersistentDispatcherMultipleConsumers#trackDelayedDelivery), but
clearDelayedMessages() doesn’t seem synchronized and
InMemoryDelayedDeliveryTracker#clear() isn’t synchronized either.
Is clear() guaranteed to be called under the same lock, or should we align
with BucketDelayedDeliveryTracker#clear() (synchronized) to avoid concurrent
access to delayedMessageMap/bitmaps?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
