lizhimins opened a new pull request, #1308:
URL: https://github.com/apache/rocketmq-clients/pull/1308
<!-- Please make sure the target branch is right. In most case, the target
branch should be `master`. -->
### Which Issue(s) This PR Fixes
Fixes #1307
### Brief Description
Port the Java graceful-shutdown fix (#992 for #816) to the C++
`PushConsumer` so that in-flight `ReceiveMessage` requests complete and cached
messages are consumed / acked before the underlying gRPC sessions are torn down.
Previously `PushConsumerImpl::shutdown()` cleared `process_queue_table_` and
force-stopped the consumption `io_context` before `ClientImpl::shutdown()`,
which caused:
- **Latency spikes** — messages already in the local cache were dropped
locally and only came back after the broker's invisible-duration expired.
- **Duplicate consumption** — messages that finished consumption during
shutdown could not `Ack` because the gRPC channel was already gone.
**Approach (aligned with Java #992):**
1. `ThreadPool` gains `gracefulShutdown()` — releases the asio
`executor_work_guard` and joins workers **without** calling `context_.stop()`,
letting queued handlers drain naturally.
2. `ConsumeMessageService` gains `gracefulShutdown()` that forwards to the
pool.
3. `PushConsumerImpl` tracks in-flight `ReceiveMessage` RPCs via
`std::atomic<int64_t>` — incremented in `ProcessQueueImpl::popMessage()` before
dispatching, decremented from the completion callback.
4. `PushConsumerImpl::shutdown()` is rewritten into six ordered phases:
1. Cancel `scan_assignment_handle_` / `collect_stats_handle_`.
2. `awaitInflightReceiveRequests()` — poll until the inflight counter
hits 0 or `request_timeout + polling_timeout` elapses.
3. `awaitCachedMessagesDrained()` — poll until every
`ProcessQueue::cachedMessageQuantity()` is 0, bounded by the same deadline.
4. `consume_message_service_->gracefulShutdown()`.
5. Sleep 1s to let async `ack` / `nack` RPCs complete.
6. Clear `process_queue_table_` and call `ClientImpl::shutdown()`.
All wait loops are strictly time-bounded so a stuck broker cannot hang
`shutdown()`.
### How Did You Test This Change?
- New unit tests in `cpp/source/rocketmq/tests/GracefulShutdownTest.cpp`:
- `inflightCounterInitiallyZero` — counter starts at 0.
- `inflightCounterIncrementDecrement` — `increment` / `decrement`
arithmetic is correct.
- `threadPoolGracefulShutdownDrainsTasks` — 10 tasks posted with 10ms
sleep all complete after `gracefulShutdown()`.
- `threadPoolGracefulShutdownRejectsNewTasks` — tasks submitted after
`gracefulShutdown()` are rejected.
- `shutdownOnNonStartedConsumerIsNoOp` — CAS guard prevents shutdown side
effects on unstarted consumers.
- Regression: `push_consumer_impl_test`, `process_queue_impl_test`,
`consume_message_service_test`, `consume_task_test` all pass.
Local verification:
```bash
cd cpp
bazel build //source/rocketmq:rocketmq_library
bazel test //source/rocketmq/tests:graceful_shutdown_test \
//source/rocketmq/tests:push_consumer_impl_test \
//source/rocketmq/tests:process_queue_impl_test \
//source/rocketmq/tests:consume_message_service_test \
//source/rocketmq/tests:consume_task_test
```
All targets build and pass.
--
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]
