[ISSUE #10827] fix(broker): spin for the lock on same-attemptId pop orderly retry to avoid empty response

[lizhimins]

What is the purpose of the change

Fixes #10827.

An orderly (fifo / PopKv) retry carrying the same attemptId is an idempotent reentrant request, but PopConsumerService.popAsync fails fast with an empty response when the group@topic lock is contended. The retry then suspends in long polling, burns the only reentrant opportunity, times out, and the client rotates to a new attemptId — permanently losing reentrancy and blocking the queue head (up to one invisibleTime, or ~3h when proxy autoRenew keeps extending nextVisibleTime). See the issue for the full failure chain, verified step by step against code in a real incident (a patrol system consuming orderly messages through Proxy intermittently hung for a long time).

Brief description of the change

Introduce tryLockForPop in PopConsumerService: a fifo request whose attemptId is already registered in OrderInfo (i.e. a genuine in-flight retry of a previous delivery of the same receive attempt, checked via the new ConsumerOrderInfoManager.isAttemptIdMatched) spin-retries tryLock until the lock is acquired instead of failing fast; all other requests keep the fail-fast behavior — pops with a different attemptId would be blocked by checkBlock even after acquiring the lock, so spinning would only waste request threads while the lock may be held for seconds on slow paths.

Correctness rationale:

• The lock holder always releases on pop completion (whenComplete unlock), with the lock service's 2-minute expiry sweep as the worst-case backstop, so the spin cannot wait forever.
• Same-attemptId contention is rare and the wait is normally milliseconds, so busy-wait cost is negligible.
• Once the lock is acquired, the existing re-pop path runs (checkBlock passes for the same attemptId, consumedCount is not incremented), fully preserving the reentrant semantics.
• The wakeUp long-polling re-execution path benefits as well.

A lock-free read-only replay of the previous OrderInfo batch was evaluated first; it works but is much larger, and further analysis showed the existing re-pop path is already self-consistent (invalidating the receipt handle of the lost delivery is normal at-least-once semantics) — the real defect is only that lock contention leaves the retry empty, hence this minimal fix (one new interface method with a small lookup, no config switch).

Does this pull request affect any existing functionality?

No. Only fifo requests whose attemptId is already registered in OrderInfo change behavior (spin instead of empty response on lock contention); all other requests keep fail-fast.

Verification

• New unit tests PopConsumerServiceLockRetryTest: spin until the lock is acquired and the pop path proceeds when the attemptId is registered; unregistered attemptId and non-fifo keep fail-fast. ConsumerOrderInfoManagerTest covers isAttemptIdMatched. All 17 existing PopConsumerServiceTest cases pass, checkstyle clean, JDK 8 compile verified.
• Real-environment verification (k8s micro setup, broker image with this fix):
  • End-to-end: first pop FOUND/2 unacked → same-attemptId re-pop FOUND/2 within milliseconds → ack succeeds → fresh-attemptId pop finds the queue drained.
  • Contention probe: 5 rounds × 16 concurrent same-attemptId pops, all 80 returned the same batch with zero empty responses; broker logs confirmed all 16 requests per round executed inside the lock, i.e. the contention window was genuinely exercised.

[RockteMQ-AI]

Summary

Reviewed the changes in this PR. Overall the changes look reasonable.

A few observations from the automated analysis:

• Please ensure adequate test coverage for the changes
• Verify backward compatibility if any public APIs are modified

---

Automated review by github-manager-bot

[codecov-commenter]

Codecov Report

❌ Patch coverage is 82.35294% with 3 lines in your changes missing coverage. Please review.
✅ Project coverage is 48.32%. Comparing base (2daf0e2) to head (43461f9).
⚠️ Report is 4 commits behind head on develop.

Files with missing lines	Patch %	Lines
...apache/rocketmq/broker/pop/PopConsumerService.java	77.77%	0 Missing and 2 partials ⚠️
.../broker/pop/orderly/QueueLevelConsumerManager.java	87.50%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             develop   #10828      +/-   ##
=============================================
- Coverage      48.32%   48.32%   -0.01%     
- Complexity     13516    13538      +22     
=============================================
  Files           1380     1380              
  Lines         101138   101165      +27     
  Branches       13120    13127       +7     
=============================================
+ Hits           48876    48886      +10     
+ Misses         46298    46292       -6     
- Partials        5964     5987      +23     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[fuyou001]

Thanks for fixing the lost-reentrancy path described in #10827. I traced the current head through POP entry, long-poll wakeup, ACK/change-invisible, lock cleanup, request lifetime, and persisted OrderInfo. The direction is reasonable, but the new synchronous wait couples several P1 correctness and availability risks, so I do not think the busy-spin implementation is safe to merge yet.

P1: the unbounded spin can exhaust the shared Pull executor and outlive the request

PopConsumerService.tryLockForPop spins at lines 493-495 before popAsync can return a Future. Network POP and long-poll wakeup execute on pullMessageExecutor, which is also shared by PULL, PEEK, POP_LITE, NOTIFICATION, and POLLING_INFO; Local Proxy invokes the processor synchronously on its caller thread. A burst of same-attemptId retries can therefore occupy the complete shared pool while continuously calling Thread.yield().

The loop has no request deadline, channel-active, interruption, cancellation, or Broker-stop check. The timeout check in PopMessageProcessor happens only before entering popAsync. Consequently a request can expire or disconnect, later acquire the lock, and still rewrite OrderInfo/popTime. Multiple stale retries can also invalidate the receipt returned by a live retry.

The premise that every holder releases is not true on all existing paths: after acquisition, a deleted or consume-disabled subscription group returns at PopConsumerService.java:367-370 before the whenComplete unlock is installed, and a synchronous setup exception can reach the catch/return at lines 466-470 with the same result.

Please replace the worker-thread busy-spin with bounded, cancellable asynchronous acquisition or a lock-release notification path. Enforce the original request deadline and channel/service state, and put every successful acquisition under one ownership guard.

P1: timeout cleanup is not a safe backstop and can break mutual exclusion

The new comment at PopConsumerService.java:488-490 relies on PopConsumerLockService.removeTimeout(). That method removes an entry after two minutes even when its locked flag is true. The spinner then creates and acquires a new TimedLock for the same key while the old holder can still be inside the critical section.

There is a second ownership problem: unlock(key) looks up the current map entry instead of releasing the instance acquired by the caller. The concrete sequence is:

1. holder A owns old lock L1;
2. cleanup removes locked L1;
3. waiter B creates and acquires L2, so A and B now overlap;
4. A executes unlock(key), finds L2, and releases the lock owned by B;
5. a third request can enter concurrently with B.

This can concurrently mutate OrderInfo and consumerOffset across POP, ACK, and ChangeInvisible. The existing lock test even verifies removal while locked=true, but does not test ownership after replacement.

Please do not use removal of a held entry as lock handoff. Fix the leak paths and only remove safely idle entries, or introduce owner/epoch/fencing semantics so an old holder cannot commit or unlock a replacement. Add a regression with old holder + timeout cleanup + waiter + old unlock + third contender, asserting that the critical-section concurrency never exceeds one.

P1: waiting consumes the invisible time before delivery

PopMessageProcessor captures beginTimeMills before calling popAsync, and PopConsumerContext stores that value as final popTime before tryLockForPop waits. After the lock is eventually acquired, the same old value is persisted into OrderInfo and returned in the receipt. FIFO blocking calculates visibility as popTime + invisibleTime.

If the wait approaches or exceeds invisibleTime, the returned message has a shortened or already-expired lease. Once the lock is released, a different attemptId can immediately pass checkBlock and read the same queue head while the first client is still processing it.

After bounded acquisition, recheck the request deadline and establish popTime at the actual delivery epoch before reading/updating OrderInfo. Add a test that holds the lock longer than invisibleTime and verifies that the returned receipt remains invisible for the full configured duration from actual delivery.

P1: a stored matching string does not prove an active retry for the target queue

QueueLevelConsumerManager.isAttemptIdMatched scans every queue under topic/group and checks only the stored string. It does not receive the requested queueId, inspect ACK bits or next-visible-time, or revalidate after acquiring the lock.

OrderInfo remains after all messages are ACKed or visibility expires, is cleaned only after up to 24 hours, and is persisted/restored across Broker restart. Therefore fully-ACKed, expired, restarted stale IDs, and an ID stored only on another queue can all enter the unbounded wait. The check is also outside the lock, so the current holder can ACK or replace the OrderInfo while the waiter is spinning.

Please make the predicate queue-aware and state-aware: for an explicit queue, require a matching active unacked delivery on that queue; only queueId=-1 should scan queues. Revalidate under the acquired lock before reading messages. Regression coverage should include all-ACKed, expired, explicit-queue mismatch, state replacement while waiting, and persisted active versus stale OrderInfo.

Suggested validation

The current tests pass, but PopConsumerServiceLockRetryTest only makes a mock tryLock succeed on call 51. It does not exercise a real holder, request expiry/cancellation, shared-executor saturation, timeout cleanup, receipt validity, or state changes while waiting. Those scenarios should be covered before relying on the new waiting path.

[lollipopjin]

LGTM

[fuyou001]

Follow-up on the P1 lock-wait issue: Thread.yield() cannot provide the progress guarantees this loop relies on.

The JDK contract defines Thread.yield() only as a scheduler hint, and the scheduler is free to ignore it. It does not park the waiter, guarantee that the current lock holder will run, provide fairness, or impose any upper bound on acquisition time. Consequently, while (!tryLock(...)) { Thread.yield(); } can keep a pullMessageExecutor worker RUNNABLE indefinitely; with multiple same-attemptId retries, the scheduler may simply rotate among spinning waiters while they continue consuming CPU. It also adds no deadline, cancellation, interruption, disconnect, or shutdown handling.

Please do not use yield as the synchronization/progress mechanism here. Prefer an explicit asynchronous unlock notification/handoff, or at minimum a bounded timed acquisition tied to the request deadline and lifecycle cancellation, followed by state revalidation after acquisition. A regression test should hold the lock beyond the request deadline and assert that the request completes without occupying the worker indefinitely and cannot mutate POP/OrderInfo state after expiration; a multi-waiter test should also verify that the shared executor is not exhausted.