[ISSUE #1340] [Java] Resume PushConsumer receive after cache drains

[qianye1001]

What is the purpose of the change

Fixes #1340.

Java PushConsumer currently waits for the full one-second cache backoff after a ProcessQueue becomes full, even when ACK/NACK completion has already drained its local cache. With many assigned queues, the per-queue cache quota can be small, so this fixed receive gap can repeat under backlog and reduce receive throughput further.

Issue #1196 and PR #1214 increased the default total message cache, but the fixed recovery gap remained unchanged.

Brief changelog

• Add a distinct CachePause token for each cache-full pause and retain that round's not-yet-sent receive attemptId.
• Store the active pause in an AtomicReference and use token-identity CAS so ACK/NACK completions and the fallback timer can schedule at most one resume.
• Resume through the consumer scheduler when both cached message count and cached bytes are at or below 20% of their current per-queue thresholds (threshold / 5L). Thresholds below five therefore use a zero low watermark.
• Keep the existing one-second task as a liveness fallback.
• Ignore stale timers from older pause rounds, preventing ABA against a newer pause.
• Stop recovery for dropped queues and stopped consumers.

The total count/byte cache limits and their existing per-queue high-watermark calculation are unchanged.

Verifying this change

Added unit coverage for:

• exact 20% count and byte low-watermark boundaries;
• zero low watermark when a threshold is below five;
• concurrent cache eviction and fallback timer scheduling only one resume;
• a stale timer not affecting a newer pause;
• reuse of the paused receive attemptId;
• dropped queues and stopped consumers.

Validation with JDK 11:

mvn -pl client -am -DskipITs -DfailIfNoTests=false -Dtest=ProcessQueueImplTest test
Tests run: 19, Failures: 0, Errors: 0, Skipped: 0

mvn -pl test -am -DskipITs -DfailIfNoTests=false -Dtest=AttemptIdIntegrationTest test
Tests run: 1, Failures: 0, Errors: 0, Skipped: 0

Both reactor builds also completed Checkstyle and SpotBugs successfully.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 82.35294% with 6 lines in your changes missing coverage. Please review.
✅ Project coverage is 62.72%. Comparing base (4076e2b) to head (9fa3c7c).
⚠️ Report is 163 commits behind head on master.

Files with missing lines	Patch %	Lines
...mq/client/java/impl/consumer/ProcessQueueImpl.java	82.35%	4 Missing and 2 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #1341      +/-   ##
============================================
+ Coverage     53.26%   62.72%   +9.45%     
- Complexity      651      828     +177     
============================================
  Files           208      108     -100     
  Lines         14303     4663    -9640     
  Branches       5845      366    -5479     
============================================
- Hits           7619     2925    -4694     
+ Misses         6308     1540    -4768     
+ Partials        376      198     -178     

Flag	Coverage Δ
cpp	?
golang	?
java	62.72% <82.35%> (+1.04%)	⬆️
rust	?

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ 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.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[RockteMQ-AI]

Summary

This PR modifies 2 file(s) (+329/-3) on branch codex/cache-full-event-resume → master.

Files changed:

• java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ProcessQueueImpl.java
• java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/ProcessQueueImplTest.java

A maintainer will provide a detailed code review. This automated review confirms the PR is structurally sound and ready for human review.

---

Automated review by github-manager-bot

[lizhimins]

这设计有点太复杂了

[RockteMQ-AI]

LGTM. The cache-pause mechanism is a solid improvement over the previous polling-based backoff.

Key strengths:

• Correct concurrency: with CAS ensures exactly-once resume even under concurrent eviction + fallback timer
• Hysteresis via low watermark (1/5 threshold): prevents oscillation between pause/resume
• Lost-wake protection: called after installing the pause covers the race where eviction happens during setup
• Lifecycle safety: clears the pause ref; checks before
• Thorough tests: 8 new test cases covering watermark boundaries, concurrency, stale fallback, drop, and consumer-stop scenarios

---

Automated review by github-manager-bot