qianye1001 opened a new issue, #1340:
URL: https://github.com/apache/rocketmq-clients/issues/1340
### Before Creating the Issue
- [x] I have searched the existing issues and did not find a matching open
issue.
### Programming Language
Java
### Version
Current `master`
### What happened
The Java gRPC PushConsumer divides its total message-count and byte cache
limits across all assigned `ProcessQueue` instances:
```text
per-queue count threshold = maxCacheMessageCount / processQueueCount
per-queue byte threshold = maxCacheMessageSizeInBytes / processQueueCount
```
When a client is assigned many queues, each queue can therefore receive only
a small cache quota. Once either quota is full,
`ProcessQueueImpl.receiveMessage` schedules the next receive attempt after a
fixed one-second delay.
ACK/NACK completion removes messages from the local cache, but cache
eviction does not wake the paused receive loop. Even if consumption drains the
queue almost immediately, that queue still waits for the one-second timer.
With server-side backlog, each receive can return immediately, so the client
may repeatedly enter this cycle:
```text
many assigned queues
-> small per-queue quota
-> cache full
-> fixed one-second receive gap
-> lower receive throughput
-> consumer lag grows
-> subsequent receives fill the cache immediately
```
This can make receive request frequency and consumption throughput fall
while lag continues to grow.
Issue #1196 proposed a per-queue cache option. PR #1214 increased the
default total message cache from 1024 to 4096, which mitigates the frequency of
cache-full events but does not remove the fixed recovery gap.
### Expected behavior
A cache-full queue should remain paused while it is near the high watermark,
but should resume promptly after ACK/NACK completion drains both cache
dimensions to a low watermark. The existing one-second task should remain only
as a fallback.
### Proposed fix
- Represent each cache-full pause with a distinct token that retains the
not-yet-sent receive `attemptId`.
- Store the active token in an `AtomicReference`.
- After cache eviction, resume when both cached message count and cached
bytes are at or below 20% of their current per-queue thresholds (`threshold /
5L`; naturally zero for thresholds below five).
- Use token-identity CAS so concurrent ACKs and the fallback timer can
schedule only one resume, and a stale timer cannot affect a newer pause (ABA).
- Resume through the consumer scheduler and reuse the paused attempt's
`attemptId`.
- Do not resume a dropped queue or a stopped consumer.
- Keep the existing one-second timer as a liveness fallback.
### Scope
This issue concerns the Java gRPC `PushConsumer` / `ProcessQueueImpl`
receive loop. `SimpleConsumer` uses a separate receive path and is not affected
by this change.
--
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]
