yx9o commented on code in PR #11434:
URL: https://github.com/apache/rocketmq/pull/11434#discussion_r4420906431
##########
broker/src/main/java/org/apache/rocketmq/broker/longpolling/PopLiteLongPollingService.java:
##########
@@ -58,7 +58,18 @@ public PopLiteLongPollingService(BrokerController
brokerController, NettyRequest
this.brokerController = brokerController;
this.processor = processor;
this.pollingMap = new ConcurrentLinkedHashMap.Builder<String,
ConcurrentSkipListSet<PopRequest>>()
-
.maximumWeightedCapacity(this.brokerController.getBrokerConfig().getPopPollingMapSize()).build();
+
.maximumWeightedCapacity(this.brokerController.getBrokerConfig().getPopPollingMapSize())
+ // A queue evicted at capacity may still hold suspended requests.
Wake them up
+ // instead of leaving them unreachable for both
notifyMessageArriving() and the
+ // timeout sweep in run().
+ .listener((key, queue) -> {
+ PopRequest request;
+ while ((request = queue.pollFirst()) != null) {
+ totalPollingNum.decrementAndGet();
+ wakeUp(request);
Review Comment:
What happens if `wakeUp()` is rejected because the pull executor is full?
The exception stops this loop, leaving the remaining requests in an evicted
queue where neither notifications nor the timeout scan can find them. Could we
keep them reachable or respond to them on rejection?
--
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]
