lizhimins opened a new issue, #1307: URL: https://github.com/apache/rocketmq-clients/issues/1307
### Before Creating the Enhancement Request - [x] I have confirmed that this should be classified as an enhancement rather than a bug/feature. ### Programming Language of the Client C++ ### Summary Align the C++ `PushConsumer` shutdown flow with the Java fix delivered in #992 (fixes #816), so that when a `PushConsumer` is being closed, the client drains all in-flight `ReceiveMessage` gRPC requests and consumes/acks the messages already cached in `ProcessQueue`s **before** the underlying gRPC connections are torn down. ### Motivation Today, `PushConsumerImpl::shutdown()` in the C++ client behaves the same way the Java client did before #992: 1. Cancel the assignment-scanning timer. 2. Clear the `process_queue_table_` immediately. 3. Force-shutdown the consumption thread pool (`context_.stop()`), abandoning queued tasks. 4. Tear down the underlying gRPC sessions via `ClientImpl::shutdown()`. Two concrete problems fall out of this ordering: - Messages already fetched into the local cache but not yet consumed are effectively dropped locally. They are only redelivered after the broker's invisible-duration expires, which shows up as a **sharp latency spike** for those messages during rolling restarts / graceful re-deployments. - Messages that finish consumption while the underlying client is being torn down cannot successfully `Ack`, because the gRPC channel is already closed. The broker will re-deliver them, causing **duplicate consumption**. Java already fixed this in commit `20757cc5` ([#992](https://github.com/apache/rocketmq-clients/pull/992), fixes [#816](https://github.com/apache/rocketmq-clients/issues/816)). The same guarantee should hold for the C++ SDK. ### Describe the Solution You'd Like Mirror the Java solution, adapted to the C++ / Asio-based execution model: 1. **Track in-flight `ReceiveMessage` requests.** Add an `std::atomic<int64_t> inflight_receive_requests_` on `PushConsumerImpl`, incremented right before `ClientManager::receiveMessage(...)` is invoked in `ProcessQueueImpl::popMessage()` and decremented from the response callback (both success and failure paths). 2. **Add `ThreadPool::gracefulShutdown()`.** A new virtual method on `ThreadPool` (implemented by `ThreadPoolImpl`) that only releases the `asio::executor_work_guard` and joins worker threads — it does NOT call `context_.stop()`, so already-posted handlers naturally drain. 3. **Add `ConsumeMessageService::gracefulShutdown()`.** Forwards to `pool_->gracefulShutdown()`. 4. **Re-order `PushConsumerImpl::shutdown()`** into six phases: 1. Cancel `scan_assignment_handle_` / `collect_stats_handle_` (no new receives, no new process queues). 2. `awaitInflightReceiveRequests()` — poll until `inflight_receive_requests_ == 0` or `request_timeout + polling_timeout` elapses. 3. `awaitCachedMessagesDrained()` — poll until sum of `cachedMessageQuantity()` across `process_queue_table_` reaches zero, bounded by the same deadline. 4. `consume_message_service_->gracefulShutdown()` — drain the consumption thread pool without discarding queued tasks. 5. Sleep 1s to let asynchronous `ack` / `nack` RPCs complete. 6. Clear `process_queue_table_` and call `ClientImpl::shutdown()` to close gRPC sessions. The wait loops are strictly time-bounded (`request_timeout + polling_timeout`) so that a stuck broker cannot hang the caller of `shutdown()`. ### Describe Alternatives You've Considered - **Do nothing and rely on the broker's invisible-duration for redelivery.** This is the current behavior. It is correct in terms of at-least-once semantics, but the observed latency spike and duplicate consumption during restarts are painful in practice, and Java already solved this — divergence between SDKs is undesirable. - **Force-flush all cached messages by rejecting the pending consume tasks on shutdown.** This would eliminate the drain wait, but keeps the duplicate-consumption problem intact (acks would still fail). It also worsens the latency issue rather than fixing it. - **Block `shutdown()` on the consumption pool via a `CountDownLatch` per message instead of polling.** Cleaner in theory, but requires threading a new synchronization primitive through `ConsumeTask` / `AsyncReceiveMessageCallback`. The polling variant matches the Java implementation directly and is trivially bounded by the same timeout, so we adopt it for consistency. ### Additional Context - Java implementation reference: commit `20757cc5` — [Fix ack fail when Java Pushconsumer shutdown (#992)](https://github.com/apache/rocketmq-clients/pull/992) - Related Java issue: [#816](https://github.com/apache/rocketmq-clients/issues/816) Happy to send the C++ PR once this enhancement is agreed on. -- 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]
