apache / apache/rocketmq-clients

[Bug] Java PushConsumer waits the full cache backoff after cache has drained

Open
#1,340 6 comments 0 reactions 0 assignees View on GitHub
Dominant language
Java
Stars
505
Forks
313
Avg merge
11h 28m
Merged PRs (30d)
6

Description

### 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.

Contributor guide

No contributing guide indexed for this repository

Research direction

Start with the Java gRPC PushConsumer receive loop and ProcessQueueImpl.receiveMessage; trace the one-second cache-full fallback and the ACK/NACK cache-eviction path. Verify how the consumer scheduler and attemptId are handled, then confirm that both cache dimensions reaching the 20% low watermark trigger one resume, while dropped queues, stopped consumers, stale timers, and concurrent completions remain safe.

Written by the indexing model from the issue text.

Assessment

Tech stack
grpc, java
Domain
distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Active
Clarity
Clearly specified
Newbie friendliness
52/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.