leizhiyuan opened a new issue, #1447: URL: https://github.com/apache/rocketmq-clients/issues/1447
### 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 Linux 6.6 x86_64, 32 cores ### RocketMQ Version of the Client/Server Client: Rust `rocketmq` 5.0.0, master `e90ece7a` Server: a mock proxy that speaks the real gRPC protocol over TCP (see the benchmark in the tracking issue); the defects are also visible by reading the code against the Java client. ### Run or Compiler Version rustc 1.97.1 (also checked with 1.74.0, the MSRV) ### Describe the Bug `MessageListener` is a synchronous `Fn(&MessageView) -> ConsumeResult`, and `PushConsumer` calls it directly inside the async consumer worker ([push_consumer.rs#L917](https://github.com/apache/rocketmq-clients/blob/e90ece7a/rust/src/push_consumer.rs#L917), FIFO [#L1074](https://github.com/apache/rocketmq-clients/blob/e90ece7a/rust/src/push_consumer.rs#L1074)). A listener that blocks (synchronous database or HTTP client, `std::thread::sleep`, heavy computation) blocks a Tokio worker thread. The consumer runs `queues × consumer_worker_count_each_queue` workers (4 per queue by default), which easily exceeds the number of runtime threads, so heartbeats, the telemetry stream, the receives of every queue and the application's own tasks are starved. There is also no way to bound how many messages are consumed concurrently. Java runs the listener on a dedicated executor of `consumptionThreadCount` threads (`PushConsumerImpl`, `PushConsumerBuilder#setConsumptionThreadCount`, default 20), separate from the gRPC and scheduler threads. ### Steps to Reproduce Runtime with 2 worker threads, push consumer with 4 queues, listener that sleeps 10 ms; measure the lateness of an unrelated 10 ms `tokio::time::sleep` loop. ### What Did You Expect to See? A blocking listener does not affect the other tasks of the runtime, and its concurrency is bounded and configurable. ### What Did You See Instead? | | `master` | fix | |---|---|---| | unrelated 10 ms timer, lateness p99 | 81.9 ms | 1.9 ms | | consumed msg/s (listener blocks 10 ms) | 188 | 1,176 | | 16 queues × 4 workers: peak concurrent listener calls | 4 (= runtime threads) | 20 (= `consumption_thread_count`) | ### Additional Context Proposed fix: run the listener through `tokio::task::spawn_blocking`, bounded by a new `PushConsumerOption::consumption_thread_count` (default 20, as in Java; 0 is rejected). Part of #1445. -- 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]
