messere1 opened a new issue, #11433: URL: https://github.com/apache/rocketmq/issues/11433
### Before Creating the Bug Report - [x] I found a bug, not just asking a question, which should be created in [GitHub Discussions](https://github.com/apache/rocketmq/discussions). - [x] I have searched the [GitHub Issues](https://github.com/apache/rocketmq/issues) and [GitHub Discussions](https://github.com/apache/rocketmq/discussions) of this repository and believe that this is not a duplicate. - [x] I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ. ### Runtime platform environment Any. ### RocketMQ version develop branch (dc70260b7) ### JDK Version Any. ### Describe the Bug `PopLiteLongPollingService` keeps suspended lite long-polling requests in per-client queues stored in `pollingMap`. Both wake-up paths — `notifyMessageArriving()` and the timeout sweep in `run()` — reach a request **only through `pollingMap`**: ```java // notifyMessageArriving() ConcurrentSkipListSet<PopRequest> remotingCommands = pollingMap.get(pollingKey); if (remotingCommands == null || remotingCommands.isEmpty()) { return false; } ``` ```java // run() timeout sweep for (Map.Entry<String, ConcurrentSkipListSet<PopRequest>> entry : pollingMap.entrySet()) { ... } ``` There are two ways a queue holding a **live, suspended** request disappears from `pollingMap` while `polling()` is adding to it: **1. Race with `cleanUnusedResource()` (introduced by #10072).** Since #10072, the periodic cleanup removes entries whose queue is *empty*: ```java pollingMap.entrySet().removeIf(entry -> CollectionUtils.isEmpty(entry.getValue())); ``` `polling()` (called on processor threads) first reads the queue and only later adds to it: ```java ConcurrentSkipListSet<PopRequest> queue = pollingMap.get(pollingKey); // (1) read if (queue == null) { ... } else { int size = queue.size(); // (2) size check if (size > ...getPopPollingSize()) { return POLLING_FULL; } } if (queue.add(request)) { // (3) add remotingCommand.setSuspended(true); ... return POLLING_SUC; // (4) client told "suspended OK" } ``` The queue is routinely empty at (1): the last pending request was just woken by the timeout sweep or by `notifyMessageArriving()`, leaving the empty queue in the map. If the service thread's `cleanUnusedResource()` runs between (1)/(2) and (3), the entry is removed from `pollingMap`, and (3) adds the request to the now-orphaned queue. `polling()` still returns `POLLING_SUC`, so the caller treats the request as suspended — but neither `notifyMessageArriving()` nor the timeout sweep can ever find it. The classic `PopLongPollingService` does not have this race because its cleanup only removes queues whose topic/group metadata is gone, never on emptiness. **2. Capacity eviction.** `pollingMap` is a `ConcurrentLinkedHashMap` with `maximumWeightedCapacity(popPollingMapSize)` (default 100000). Neither POLLING_FULL check (`maxPopPollingSize`, `popPollingSize`) bounds the number of *distinct keys*, so once the map is at capacity, inserting a new client's queue silently evicts an LRU entry — even when its queue still holds suspended requests — with the same unreachable-request consequence. Impact: a suspended lite pop request is never answered by the broker — no message notification and no broker-side `POLLING_TIMEOUT`. The client blocks until its own (typically much longer) remoting invoke timeout, then sees a timeout error and re-polls. No message is lost (offsets are unaffected); this is a latency/availability defect of the lite polling path. ### Steps to Reproduce Deterministic unit-level reproduction in `PopLiteLongPollingServiceTest` (JUnit 4 + Mockito, same harness as the existing tests): 1. Seed the routine post-sweep state: an **empty** queue for `clientId` in `pollingMap` (reflection field access, as the existing tests already do). 2. Gate the interleaving: override `BrokerConfig#getPopPollingSize()` (called by `polling()` between the `get()` and the `add()`) to block on a latch. 3. Start a thread calling `polling(ctx, cmd, now, 10000, clientId, group)`; wait until it parks inside the size check. 4. On the test thread, invoke `cleanUnusedResource()` via reflection — the empty entry is removed from `pollingMap`. 5. Release the latch; `polling()` completes and returns `POLLING_SUC`. 6. Assert the invariant: the suspended request must stay reachable through `pollingMap` (`assertSame(queue, pollingMap.get(clientId))`, then `notifyMessageArriving()` must return `true`). For the eviction path: build the service with `popPollingMapSize = 1`, poll for `clientA` (POLLING_SUC), then poll for `clientB` — `clientA`'s queue is evicted while holding its suspended request, and no wake-up is ever submitted for it. ### What Did You Expect to See? Every request for which `polling()` returns `POLLING_SUC` stays reachable through `pollingMap`, so it is woken by `notifyMessageArriving()` on message arrival or answered with `POLLING_TIMEOUT` by the sweep in `run()` when its poll time elapses. ### What Did You See Instead? `polling()` returns `POLLING_SUC` while the request lives in a queue that is no longer referenced by `pollingMap`: - `notifyMessageArriving(clientId, ...)` returns `false` (map lookup misses), - the timeout sweep in `run()` never sees the queue, so the request hangs until the client-side remoting timeout fires. ### Additional Context The empty-queue removal was introduced by #10072 (which fixed a different bug: cleaning up polling structures of unsubscribed clients without returning results). The race is a narrow-window regression of that simplification; the eviction path additionally requires more than `popPollingMapSize` distinct polling clientIds. The proposed fix keeps #10072's cleanup semantics: re-establish the mapping after a successful `add()` in `polling()` (failing the polling attempt fast if another poller has meanwhile re-created the queue), and wake up the requests of a queue that is evicted at capacity. -- 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]
