broker/src/main/java/org/apache/rocketmq/broker/lite/LmqPrefixIndex.java:85 - prefix index read lock is held while running caller callbacks (blocks the store dispatch thread, self-deadlocks on any callback that mutates the index)

[Metastarx]

Which Issue(s) This PR Fixes

(本地扫描，无链接)

• Fixes #issue_id

Brief Description

缺陷说明：LmqPrefixIndex.forEachLmqByPrefix (LmqPrefixIndex.java:78-96) takes rwLock.readLock() at line 85 and then runs the caller's visitor on line 90 while the lock is still held; the lock is only released in the finally block after the whole traversal. The visitor is caller code that does arbitrary, unbounded work. Proof from the code: AbstractLiteLifecycleManager.forEachLiteTopicByPrefix:154-163 passes a lambda that calls getMaxOffsetInQueue(lmqName) per entry, and the two consumers of that path run heavy logic inside the callback: LiteEventDispatcher.doFullDispatchForWildcardGroup:272-284 performs ConsumerOffsetManager.queryOffset() (LiteEventDispatcher.isFullyConsumed:291-297, which goes to config storage for ConsumerOffsetManagerV2), subscription lookups, event-queue offers and long-poll wakeups for every matched lmq. A wildcard group matches every lmq of a parent topic, so the read lock can be held for the entire scan of a parent topic (the intended scale of lite topics is millions of lmqs), not for a short copy of a few keys. Consequences, both of which are wrong: (1) onLmqCreate:94-95 and onLmqDelete:101-102 need the write lock, and onLmqCreate is invoked from LiteEventDispatcher.dispatch:96-97, which is called by NotifyMessageArrivingListener.arriving:45, which runs on the store's ReputMessageService thread (DefaultMessageStore.notifyMessageArriveIfNecessary:2644-2651 and the lmq multi-dispatch path at 2813-2815). While a wildcard full dispatch is iterating, the reput thread blocks on the write lock, so consume-queue dispatch, long-poll notification and pop triggers stop advancing for every topic in that broker; the hold time is proportional to the number of lmqs of the parent topic and includes per-lmq offset-store reads. Because ReentrantReadWriteLock is non-fair by default, a stream of prefix scans can additionally starve the writer. (2) The lock is not upgradable: any callback that transitively calls add()/remove() on the same thread requests the write lock while holding the read lock and blocks forever. This hazard is real in this code base, not theoretical: cleanByParentTopic:224-230 has an explicit collect-then-delete workaround with the comment "forEachLiteTopicByParent and deleteLmq each hold a lock, nesting causes deadlock", cleanExpiredLiteTopic:196-207 calls deleteLmq (which ends in onLmqDelete -> LmqPrefixIndex.remove) directly from inside a visitor and is only safe because that particular visitor iterates the non-index forEachLiteTopic, and the javadoc at AbstractLiteLifecycleManager:139 and :152 plus the class contract of LmqPrefixIndex encode the restriction "caller must NOT add/remove lmqPrefixIndex inside the callback". The invariant is therefore enforced only by convention across several call sites, and violating it hangs the broker instead of failing.

复现步骤：Deterministic: build an LmqPrefixIndex, add two lmq names under one parent topic, then call forEachLmqByPrefix(prefix, name -> { index.remove(name); return true; }) from a single thread; the call never returns (the visitor tries to upgrade the read lock to the write lock). Same class of hang for any callback that calls onLmqCreate/onLmqDelete, directly or through deleteLmq. Load-related: start a broker with a lite parent topic holding many lmqs, register a wildcard lite group, and trigger a full dispatch (RequestCode.TRIGGER_LITE_DISPATCH without clientId, or the periodic full dispatch from LiteEventDispatcher.scan:400-411); while the scan runs, inspect the reput thread (jstack) and the broker log for message-arriving dispatch: ReputMessageService is parked in LmqPrefixIndex.add waiting for the write lock, and dispatchBehindBytes/dispatchBehindMilliseconds grow because no new lmq can be indexed and the reput loop is stalled.

修复方向：Stop holding the index lock across caller code; make the traversal snapshot-based. (1) LmqPrefixIndex: keep the read lock only long enough to copy the matching keys, e.g. add List<String> snapshotByPrefix(String prefix) (returns Collections.emptyList() for empty/null prefix, copies trie.prefixMap(prefix).keySet() into an ArrayList under readLock) and reimplement forEachLmqByPrefix on top of it so the visitor is invoked after the lock is released; if the visitor returns false the loop breaks as before and the method returns false, otherwise true. Run add()/remove() under a short write-lock section only. Update the class javadoc to state that callbacks never run under the lock. (2) AbstractLiteLifecycleManager.forEachLiteTopicByPrefix:154-163 iterates the snapshot, so the per-entry getMaxOffsetInQueue(lmqName) (a consume-queue lookup) and the user callback both run outside the index lock; drop the now-unnecessary "caller must NOT add/remove lmqPrefixIndex inside the callback" notes at lines 139 and 152, and simplify cleanByParentTopic:224-230 by deleting the collect-then-delete workaround and its deadlock comment, deleting directly in the visitor (keeping the iteration over the snapshot so removal during traversal is safe); add comments explaining why nesting is now safe. (3) Regression tests in broker/src/test/java/org/apache/rocketmq/broker/lite/LmqPrefixIndexTest.java following the existing JUnit4 + assert style: a test that runs forEachLmqByPrefix with a visitor that calls index.remove(name)/add(...) and asserts completion within a bounded time (ExecutorService.submit + future.get(5, SECONDS)) - this currently deadlocks; a test that starts a slow visitor (CountDownLatch inside the callback) and asserts a concurrent add() returns while the visitor is still running, proving the write lock is not held by the traversal; tests that early break still returns false, that null/empty prefix still returns false and visits nothing, and that mutating the index inside the callback does not disturb the in-flight traversal (snapshot independence). Extend AbstractLiteLifecycleManagerTest/LiteLifecycleManagerTest with a test where the forEachLiteTopicByParent visitor deletes another lmq of the same parent and asserts the call returns and only the intended lmqs are removed, and keep the existing count/collect assertions that exercise the skipped maxOffset <= 0 branch.

How Did You Test This Change?

PASSED: wsl -e bash scripts/gate-rocketmq.sh

补充说明

背景

缺陷说明：LmqPrefixIndex.forEachLmqByPrefix (LmqPrefixIndex.java:78-96) takes rwLock.readLock() at line 85 and then runs the caller's visitor on line 90 while the lock is still held; the lock is only released in the finally block after the whole traversal. The visitor is caller code that does arbitrary, unbounded work. Proof from the code: AbstractLiteLifecycleManager.forEachLiteTopicByPrefix:154-163 passes a lambda that calls getMaxOffsetInQueue(lmqName) per entry, and the two consumers of that path run heavy logic inside the callback: LiteEventDispatcher.doFullDispatchForWildcardGroup:272-284 performs ConsumerOffsetManager.queryOffset() (LiteEventDispatcher.isFullyConsumed:291-297, which goes to config storage for ConsumerOffsetManagerV2), subscription lookups, event-queue offers and long-poll wakeups for every matched lmq. A wildcard group matches every lmq of a parent topic, so the read lock can be held for the entire scan of a parent topic (the intended scale of lite topics is millions of lmqs), not for a short copy of a few keys. Consequences, both of which are wrong: (1) onLmqCreate:94-95 and onLmqDelete:101-102 need the write lock, and onLmqCreate is invoked from LiteEventDispatcher.dispatch:96-97, which is called by NotifyMessageArrivingListener.arriving:45, which runs on the store's ReputMessageService thread (DefaultMessageStore.notifyMessageArriveIfNecessary:2644-2651 and the lmq multi-dispatch path at 2813-2815). While a wildcard full dispatch is iterating, the reput thread blocks on the write lock, so consume-queue dispatch, long-poll notification and pop triggers stop advancing for every topic in that broker; the hold time is proportional to the number of lmqs of the parent topic and includes per-lmq offset-store reads. Because ReentrantReadWriteLock is non-fair by default, a stream of prefix scans can additionally starve the writer. (2) The lock is not upgradable: any callback that transitively calls add()/remove() on the same thread requests the write lock while holding the read lock and blocks forever. This hazard is real in this code base, not theoretical: cleanByParentTopic:224-230 has an explicit collect-then-delete workaround with the comment "forEachLiteTopicByParent and deleteLmq each hold a lock, nesting causes deadlock", cleanExpiredLiteTopic:196-207 calls deleteLmq (which ends in onLmqDelete -> LmqPrefixIndex.remove) directly from inside a visitor and is only safe because that particular visitor iterates the non-index forEachLiteTopic, and the javadoc at AbstractLiteLifecycleManager:139 and :152 plus the class contract of LmqPrefixIndex encode the restriction "caller must NOT add/remove lmqPrefixIndex inside the callback". The invariant is therefore enforced only by convention across several call sites, and violating it hangs the broker instead of failing.

复现步骤：Deterministic: build an LmqPrefixIndex, add two lmq names under one parent topic, then call forEachLmqByPrefix(prefix, name -> { index.remove(name); return true; }) from a single thread; the call never returns (the visitor tries to upgrade the read lock to the write lock). Same class of hang for any callback that calls onLmqCreate/onLmqDelete, directly or through deleteLmq. Load-related: start a broker with a lite parent topic holding many lmqs, register a wildcard lite group, and trigger a full dispatch (RequestCode.TRIGGER_LITE_DISPATCH without clientId, or the periodic full dispatch from LiteEventDispatcher.scan:400-411); while the scan runs, inspect the reput thread (jstack) and the broker log for message-arriving dispatch: ReputMessageService is parked in LmqPrefixIndex.add waiting for the write lock, and dispatchBehindBytes/dispatchBehindMilliseconds grow because no new lmq can be indexed and the reput loop is stalled.

修复方向：Stop holding the index lock across caller code; make the traversal snapshot-based. (1) LmqPrefixIndex: keep the read lock only long enough to copy the matching keys, e.g. add List<String> snapshotByPrefix(String prefix) (returns Collections.emptyList() for empty/null prefix, copies trie.prefixMap(prefix).keySet() into an ArrayList under readLock) and reimplement forEachLmqByPrefix on top of it so the visitor is invoked after the lock is released; if the visitor returns false the loop breaks as before and the method returns false, otherwise true. Run add()/remove() under a short write-lock section only. Update the class javadoc to state that callbacks never run under the lock. (2) AbstractLiteLifecycleManager.forEachLiteTopicByPrefix:154-163 iterates the snapshot, so the per-entry getMaxOffsetInQueue(lmqName) (a consume-queue lookup) and the user callback both run outside the index lock; drop the now-unnecessary "caller must NOT add/remove lmqPrefixIndex inside the callback" notes at lines 139 and 152, and simplify cleanByParentTopic:224-230 by deleting the collect-then-delete workaround and its deadlock comment, deleting directly in the visitor (keeping the iteration over the snapshot so removal during traversal is safe); add comments explaining why nesting is now safe. (3) Regression tests in broker/src/test/java/org/apache/rocketmq/broker/lite/LmqPrefixIndexTest.java following the existing JUnit4 + assert style: a test that runs forEachLmqByPrefix with a visitor that calls index.remove(name)/add(...) and asserts completion within a bounded time (ExecutorService.submit + future.get(5, SECONDS)) - this currently deadlocks; a test that starts a slow visitor (CountDownLatch inside the callback) and asserts a concurrent add() returns while the visitor is still running, proving the write lock is not held by the traversal; tests that early break still returns false, that null/empty prefix still returns false and visits nothing, and that mutating the index inside the callback does not disturb the in-flight traversal (snapshot independence). Extend AbstractLiteLifecycleManagerTest/LiteLifecycleManagerTest with a test where the forEachLiteTopicByParent visitor deletes another lmq of the same parent and asserts the call returns and only the intended lmqs are removed, and keep the existing count/collect assertions that exercise the skipped maxOffset <= 0 branch.

变更内容

.../broker/lite/AbstractLiteLifecycleManager.java  |  35 +++--
 .../rocketmq/broker/lite/LmqPrefixIndex.java       |  50 +++++--
 .../lite/AbstractLiteLifecycleManagerTest.java     |  64 ++++++++
 .../broker/lite/LiteLifecycleManagerTest.java      |  50 +++++++
 .../rocketmq/broker/lite/LmqPrefixIndexTest.java   | 162 ++++++++++++++++++++-
 5 files changed, 338 insertions(+), 23 deletions(-)

验证

PASSED: wsl -e bash scripts/gate-rocketmq.sh

检查清单

• 已通过本地测试
• 变更最小且不包含无关文件
• 已遵守仓库贡献规范

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.