leizhiyuan opened a new issue, #1422: URL: https://github.com/apache/rocketmq-clients/issues/1422
### 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](https://github.com/apache/rocketmq-clients/issues) and [GitHub Discussions](https://github.com/apache/rocketmq-clients/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 Rust ### Runtime Platform Environment macOS 27 (arm64) ### RocketMQ Version of the Client/Server Client: Rust `rocketmq` 5.0.0, master `a3d1de8b` with #1404 and #1381 merged on top (both still open) Server: a managed RocketMQ 5.x service (Tencent Cloud TDMQ for RocketMQ 5), gRPC proxy endpoint ### Run or Compiler Version rustc 1.98.0 ### Describe the Bug `PushConsumer::shutdown()` / `LitePushConsumer::shutdown()` takes several minutes. The time grows with the number of assigned queues. **Scenario.** A `LitePushConsumer` with one lite subscription is assigned 6 queues, one per broker. It sits idle for 35 s and then calls `shutdown()`: | | shutdown time | |---|---| | Rust, `master` + #1404 + #1381 | 208.7 s (log: `shutdown 6 actors`, then the actors finish one by one, 25 to 40 s apart) | | Rust, `master` | 20 to 345 s across runs | | Java 5.2.2, same server / group / topic | about 6 s | The background task that keeps re-creating sessions after shutdown is a separate problem. #1381 fixes it (`LiteSubscriptionManager::stop_scheduler`). This issue is only about the time `shutdown()` itself takes. **Root cause.** Two things in `rust/src/push_consumer.rs` add up: 1. **Workers keep receiving after shutdown starts.** Each `MessageQueueActor` worker runs ```rust loop { select! { _ = ticker.tick() => { consumer_worker.receive_messages(...).await ... } _ = shutdown_token.cancelled() => { ...; break; } } } ``` The ticker fires every 1 ms, so it is always ready again by the time a long polling request returns. `select!` picks a ready branch at random, so after shutdown starts the worker often issues another long poll instead of stopping. Each extra poll can take up to `long_polling_timeout` + `timeout`. 2. **Actors are shut down one by one.** `PushConsumer` (in both `start()` and `start_with_telemetry()`) and `process_assignments` do `for (_, actor) in entries { actor.shutdown().await }`. Each actor waits for its own in-flight long poll, so the waits add up across queues. **Java behavior.** - `ProcessQueueImpl#receiveMessage` checks `dropped` and `consumer.isRunning()` before every receive request, so no new request is issued once shutdown starts. - `PushConsumerImpl#shutDown` cancels the assignment scan, then calls `waitingReceiveRequestFinished()` once for all queues. That waits for the in-flight receive requests to finish, bounded by `requestTimeout + longPollingTimeout` in total, not per queue. ### Steps to Reproduce 1. Start a `LitePushConsumer` (or `PushConsumer`) on a topic with several queues, and `subscribe_lite` one lite topic. 2. Wait until the workers are long polling (e.g. 35 s). 3. Measure `shutdown().await`. ### What Did You Expect to See? Shutdown takes at most about one long polling timeout, like the Java client, no matter how many queues are assigned. ### What Did You See Instead? 208.7 s for 6 queues, with the actors finishing one after another. ### Additional Context The fix depends on #1404 and #1381, which touch the same code. It is based on a branch with both merged, and its PR should be merged after them. --- Tracked in #1412. -- 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]
