leizhiyuan opened a new issue, #1446: URL: https://github.com/apache/rocketmq-clients/issues/1446
### 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 `PushConsumer` acks every message synchronously inside the consumer worker: the next message is consumed only after the ack RPC of the previous one returned, and the next receive is sent only after the whole batch was acked ([push_consumer.rs#L917-L920](https://github.com/apache/rocketmq-clients/blob/e90ece7a/rust/src/push_consumer.rs#L917-L920)). ```rust let consume_result = consume_message(&self.message_listener, &message); match consume_result { ConsumeResult::SUCCESS => { ack_processor.ack_message(message).await; // waits for the ack round trip ``` A worker is therefore capped at `1 / (consume time + ack RTT)`. With the default 4 workers per queue and a 2 ms RTT, one queue cannot go above ~2,000 msg/s even with an empty listener. Java does not wait: `StandardConsumeService#consume` submits every message to the consumption executor, `ProcessQueueImpl#onReceiveMessageResult` issues the next receive right away, and `ProcessQueueImpl#eraseMessage` sends the ack asynchronously. Java also sends one ack per message, so batching is not the gap; the ack round trip being on the consumption path is. FIFO is not affected on purpose: `FifoConsumeService` consumes the next message of a group only after the ack of the previous one completed, which keeps the order. ### Steps to Reproduce Push consumer, 4 queues × 4 workers, batch 32, ack and receive RTT 2 ms, listener returns `SUCCESS` immediately; measure consumed messages per second. ### What Did You Expect to See? Consumption not bounded by the ack RTT, like Java. ### What Did You See Instead? 3,395 msg/s on `master`, 28,717 msg/s with the fix (end-to-end benchmark in #1445). ### Additional Context 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]
