apache / apache/rocketmq-clients
[Bug] Inflight receive request count leaks on MESSAGE_NOT_FOUND, PushConsumer GracefulStop always waits full requestTimeout + longPollingTimeout (Go SDK v5.1.4+)
- Dominant language
- Java
- Stars
- 505
- Forks
- 313
- Avg merge
- 11h 28m
- Merged PRs (30d)
- 6
Description
### Before Creating the Bug Report
- [x] I found a bug, not just asking a question, which should be created in [GitHub Discussions](https://github.com/apache/rocketmq-clients/discussions).
- [x] I have searched the GitHub Issues and GitHub Discussions of this repository and believe that this is not a duplicate.
- [x] I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ.
### Programming Language of the Client
Go
### Runtime Platform Environment
OS: Windows 11 25H2, amd64
RocketMQ server: 5.x, proxy mode (gRPC). longPollingTimeout is delivered to the client via syncSettings (20s in our environment; the default ProxyConfig grpcClientConsumerMaxLongPollingTimeoutMillis).
### RocketMQ Version of the Client/Server
Client: v5.1.4 (github.com/apache/rocketmq-clients/golang/v5 v5.1.4)
Server: RocketMQ 5.x (proxy mode)
Regression introduced in: PR #1253 (fix log format verbs, merged 2026-06-15)
### Run or Compiler Version
Compiler: go1.26.5 windows/amd64
### Describe the Bug
When a PushConsumer receives MESSAGE_NOT_FOUND (long-polling expired with no new message), the isNoNewMessage branch in process_queue.go only logs at debug level and **does not call cli.doAfter(MessageHookPoints_RECEIVE, ...)**. The inflight counter inflightReceiveRequestCount (maintained by defultInflightRequestCountInterceptor in push_consumer.go) is therefore never decremented.
As a result, GracefulStop() -> waitingReceiveRequestFinished() can never observe a zero counter and always waits the full timeout
equestTimeout + longPollingTimeout (3s + 20s = 23s in our environment), then logs:
`
Timeout waiting for all inflight receive requests to be finished, inflightReceiveRequestCount=11
`
Shutdown takes a **fixed ~24s** (23s + 1s final sleep) on every restart, independent of whether any request is actually in flight.
### Steps to Reproduce
1. Start a PushConsumer (Go SDK v5.1.4) subscribed to a topic with no incoming messages, so long-polling requests keep expiring with MESSAGE_NOT_FOUND.
2. Let it idle for a few minutes (each long-polling expiry leaks one inflight count).
3. Call GracefulStop() (or shut down the process).
4. Observe: after Waiting for the inflight receive requests to be finished, it always waits the full
equestTimeout + longPollingTimeout, then logs Timeout waiting for all inflight receive requests to be finished, inflightReceiveRequestCount=N (N grows with idle time).
### What Did You Expect to See?
GracefulStop() should return as soon as the single real in-flight long-polling request finishes (0 to longPollingTimeout), like v5.1.3 where every error path calls doAfter and the counter is correctly released.
### What Did You See Instead?
A fixed ~24s shutdown (3s + 20s + 1s with our 20s longPollingTimeout) on every restart. SDK log timeline:
`
12:53:30.360 Waiting for the inflight receive requests to be finished
12:53:53.459 WARN Timeout waiting for all inflight receive requests to be finished, inflightReceiveRequestCount=11
12:53:54.459 start notifyClientTermination
`
### Additional Context
Root cause, golang/process_queue.go (
eceiveMessageImmediatelyWithAttemptId), identical in v5.1.4 and master:
`go
rpcErr, isRpcErr := AsErrRpcStatus(err)
isNoNewMessage := isRpcErr && rpcErr.GetCode() == int32(v2.Code_MESSAGE_NOT_FOUND)
if isNoNewMessage {
dpq.consumer.cli.log.Debugf("No new message, mq=%s, endpoints=%v, clientId=%s",
dpq.mqstr, endpoints, clientId)
} else {
dpq.consumer.cli.doAfter(MessageHookPoints_RECEIVE, make([]*MessageCommon, 0), duration, MessageHookPointsStatus_ERROR)
dpq.consumer.cli.log.Errorf("Exception raised during message reception, ...")
}
`
The isNoNewMessage branch was added in PR #1253 to downgrade the noisy ERROR log on normal long-polling expiry, but it forgot to call cli.doAfter, leaking the inflight counter. v5.1.3 and earlier have no such branch: every error path calls doAfter(..., MessageHookPointsStatus_ERROR) and there is no leak (the trade-off there is a noisy ERROR log every expiry). The bug is still present on master (verified byte-identical to v5.1.4 as of 2026-08-06).
Suggested fix (one line, use OK status since MESSAGE_NOT_FOUND is a normal no-message result):
`go
if isNoNewMessage {
dpq.consumer.cli.doAfter(MessageHookPoints_RECEIVE, make([]*MessageCommon, 0), duration, MessageHookPointsStatus_OK)
dpq.consumer.cli.log.Debugf("No new message, mq=%s, endpoints=%v, clientId=%s",
dpq.mqstr, endpoints, clientId)
}
`
Related facts verified in the repo: Code_MESSAGE_NOT_FOUND = 40401 (protocol/v2/definition.pb.go), MessageHookPointsStatus_OK/ERROR (message.go).
Contributor guide
No contributing guide indexed for this repository
Research direction
Start in golang/process_queue.go at receiveMessageImmediatelyWithAttemptId and compare the isNoNewMessage path with the error path; then read defultInflightRequestCountInterceptor in push_consumer.go and the doAfter hook in message.go. Reproduce repeated MESSAGE_NOT_FOUND responses and call GracefulStop; done means the inflight count returns to zero and shutdown no longer waits the full requestTimeout plus longPollingTimeout.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- go, grpc
- Domain
- distributed-systems
- Issue type
- Bug
- Difficulty
- 2/5
- Estimated time
- 1-3 hours
- Activity status
- Active
- Clarity
- Clearly specified
- Newbie friendliness
- 86/100