[ISSUE #10754] Fix LMQ hold request cleanup race

[ai-yang]

Which Issue(s) This PR Fixes

Fixes #10754

Brief Description

LMQ empty-bucket cleanup previously checked ManyPullRequest.isEmpty() and then removed the map entry in separate operations. A concurrent suspend or notification replay could append a request to the selected bucket after the empty check, while cleanup still removed that bucket from pullRequestTable. The request then remained in a detached object and could no longer be reached by message arrival or timeout scans.

This change keeps bucket selection, append, and empty cleanup atomic per key:

• suspended requests and notification replays use one map-aware compute() helper;
• LMQ cleanup uses computeIfPresent() and performs the authoritative synchronized isEmpty() check inside the remapping boundary;
• ordinary non-LMQ hold-request behavior remains unchanged.

How Did You Test This Change?

• Unmodified develop: the deterministic cleanup-vs-suspend regression failed in 5/5 isolated JDK 8 Maven processes.
• Fixed targeted test class: 3/3 once, then 20 isolated Maven/JVM processes at 3/3 each (60/60 total).
• Full broker -am test: all 10 reactor modules passed; 2,369 tests, 9 skips, 0 failures, and 0 errors. Broker ran 755 tests, including the new class at 3/3.
• Checkstyle: 0 violations across all 10 modules.
• SpotBugs: 0 bugs/errors across all code modules.
• git diff --check: passed.

[RockteMQ-AI]

Review by github-manager-bot

Summary

Fixes a race condition in LMQ pull request hold service cleanup and refactors PullRequestHoldService to use atomic ConcurrentHashMap.compute() operations instead of get-then-put patterns.

Findings

• [Info] LmqPullRequestHoldService.java:57 — Replacing the unsafe get-null-check-remove pattern with computeIfPresent is the correct fix. The original code had a TOCTOU race where a new request could be added between the emptiness check and the removal.
• [Info] PullRequestHoldService.java:48-56 — The addPullRequest helper using compute() ensures atomic bucket creation and request addition. The Consumer<ManyPullRequest> parameter is a clean abstraction.
• [Info] PullRequestHoldService.java:180 — The replay list path now also uses the same addPullRequest helper, ensuring consistency.
• [Info] LmqPullRequestHoldServiceTest.java — New test file with concurrent scenario coverage using CountDownLatch for deterministic race testing. Good approach.

Suggestions

• The Consumer<ManyPullRequest> allocation per call is minor but could be replaced with a method reference or lambda cached in a field if GC pressure becomes a concern under high throughput. Not blocking.

---

Automated review by github-manager-bot