Metastarx opened a new pull request, #11219:
URL: https://github.com/apache/rocketmq/pull/11219
### 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 top
ics 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 req
uests 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
<details><summary>补充说明</summary>
## 背景
缺陷说明：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 top
ics 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 req
uests 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.
## 变更内容
```diff
.../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
## 检查清单
- [x] 已通过本地测试
- [x] 变更最小且不包含无关文件
- [x] 已遵守仓库贡献规范
</details>
--
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]
