leizhiyuan opened a new issue, #1445: URL: https://github.com/apache/rocketmq-clients/issues/1445
### 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 Rust ### Summary Tracking issue for performance problems on the hot paths of the Rust client (`PushConsumer`, the session layer shared by every client). Each problem has its own issue and PR so that they can be reviewed and merged one by one; this issue keeps the overview, the end-to-end benchmark and the suggested merge order. | # | Issue | PR | Problem | Java reference | |---|---|---|---|---| | 1 | ISSUE_1 | PR_1 | `PushConsumer` waits for every ack round trip before consuming the next message | `StandardConsumeService`, `ProcessQueueImpl#eraseMessage` (async ack) | | 2 | ISSUE_2 | PR_2 | The synchronous `MessageListener` runs on Tokio worker threads; no `consumptionThreadCount` | `PushConsumerImpl` consumption executor, `PushConsumerBuilder#setConsumptionThreadCount` | | 3 | ISSUE_3 | PR_3 | One `tokio::Mutex` over the session map is held across connect / telemetry / settings I/O | `ClientManagerImpl#getRpcClient`, `ClientImpl#syncSettings` | | 4 | ISSUE_4 | PR_4 | Ack retry queue: head-of-line blocking, `INVALID_RECEIPT_HANDLE` retried forever, unbounded | `ProcessQueueImpl#ackMessage` / `#ackMessageLater` | | 5a | ISSUE_5A | PR_5A | Rebalance starts new queues only after removed queues finish their long polling | `PushConsumerImpl#dropProcessQueue` | | 5b | ISSUE_5B | PR_5B | With several subscribed topics, every assignment scan tears down the other topics' queues | `PushConsumerImpl#syncProcessQueue` | ### Motivation Found by a code review of the Rust client against the Java client, then reproduced with unit tests and an end-to-end benchmark. Items 4 and 5b are also correctness problems (duplicate consumption, consumption interrupted every 30 s). **End-to-end benchmark.** A mock proxy speaks the real gRPC protocol over TCP on 127.0.0.1; the client is driven only through its public API (`PushConsumer`, `Producer`), release build, 32-core Linux. Every column is the same harness built against a different branch, based on `master` `e90ece7a`. 3 runs per cell, median shown. Absolute numbers depend on the mock; the ratios are the point. | Scenario | Metric | `master` | the PR alone | all PRs | |---|---|---|---|---| | 1. 4 queues × 4 workers, ack and receive RTT 2 ms, listener ~0 | consumed msg/s | 3,395 | **28,717** | **24,874** | | 2. 2 runtime threads, listener blocks 10 ms | consumed msg/s | 188 | **1,176** | **1,545** | | | lateness of an unrelated 10 ms timer, p99 | 81.9 ms | **1.9 ms** | **1.8 ms** | | 2'. 16 queues × 4 workers, listener blocks 10 ms, default options | peak concurrent listener calls / msg/s | 4 (= runtime threads) / 392 | **20** (= `consumption_thread_count`) / **1,970** | 20 / 1,968 | | 3. 8 senders to a healthy broker; one send to a new broker whose telemetry answers after 3 s | healthy sends during those 3 s | 47 | **59,966** | **55,139** | | | healthy send latency during those 3 s, p99 | 3,002 ms | **0.62 ms** | **0.63 ms** | | 4. 400 messages; first ack of each fails once; 4 receipt handles invalid | time until the 396 valid acks are done | never (140–176 after 15 s) | **1.5 s** | **1.5 s** | | | ack attempts on the 4 invalid handles in 15 s | 377–413 | **4** | **4** | | 5a. Queue 0 in a 10 s long poll, the assignment moves to queue 1 | first receive on queue 1 after it is assigned | 10,040 ms | **1 ms** | **2 ms** | | 5b. Topics a and b, one queue each, 75 s | messages consumed a / b | 0 / 373,790 | **609,013 / 608,948** | **943,392 / 943,904** | - Every PR only moves its own scenario; the other PRs stay at the `master` level in it. - In scenario 1, acked msg/s equals consumed msg/s within 0.1% for every variant: no ack is lost by sending acks in the background. - "all PRs" is ~13% below PR 1 alone in scenario 1 because the listener now runs on the blocking pool (PR 2), one thread hop per message. It trades that for the 40× lower timer latency of scenario 2. - In 5b on `master`, which topic survives varies between runs (the last one in `HashMap` order). ### Describe the Solution You'd Like One PR per item, each with regression tests that fail on `master`. Every PR passes `cargo test`, `cargo clippy --all-features -- -D warnings`, `cargo fmt --check`, `cargo doc`, `cargo build --examples`, and `cargo +1.74.0 check --all-features --tests --examples` (MSRV). **Merge order and conflicts.** Checked with `git merge-tree`: - Independent of every other PR here: PR_3. - PR_5A and PR_5B touch the same scan loop (textual conflict). PR_2 and PR_5B: no textual conflict, but PR_5B's new `scan_assignments` takes the listener type that PR_2 changes (one-line follow-up). - PR_1 conflicts with PR_2 (the worker's consume call) and with PR_4 (`AckEntryProcessor`). - Suggested order: PR_5B → PR_5A → PR_3 → PR_4 → PR_2 → PR_1. I will rebase the later PRs after each merge. I resolved the whole stack once (all 6 merged on `master`: 165 unit tests pass) to make sure the combination works, which is the "all PRs" column above. - All 6 merge cleanly with my other open Rust PR #1437. ### Describe Alternatives You've Considered One PR with all the changes: easier to merge in one go, but much harder to review, and items 4 and 5b are bug fixes that should not wait for the performance changes. ### Additional Context - Related tracking issue for Rust correctness fixes: #1412. - The benchmark harness (mock proxy + scenarios, ~700 lines) lives outside this repository. I can contribute it, e.g. under `rust/benches`, if the maintainers want it. -- 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]
