leizhiyuan opened a new pull request, #1407: URL: https://github.com/apache/rocketmq-clients/pull/1407
### Which Issue(s) This PR Fixes Fixes #1406 ### Brief Description `PushConsumer::receive_messages` looked up the filter expression by the assigned message queue's topic name (`option.get_filter_expression(&message_queue.topic.name)`). Some servers return a decorated name in `QueryAssignmentResponse`, e.g. `rmq-xxxxxxxx%my_topic`, while the subscription table is keyed by `my_topic`. The lookup always failed with `no filter expression presents`, so no `ReceiveMessage` request was ever sent. Because the worker is driven by a 1 ms ticker and had no backoff on failure, every worker spun at ~1 kHz. In one run that produced 8.94M error lines (~1.5 GB of logs) in 8 minutes, with zero messages consumed. This PR aligns the Rust `PushConsumer` with the Java client: - **Filter expression comes from the subscription, not the assigned queue.** `scanAssignments` now iterates `(topic, filter_expression)` and passes the expression to `process_assignments` → `MessageQueueActor::new`. The actor stores it and hands it to the workers. This mirrors `PushConsumerImpl#scanAssignments` → `syncProcessQueue` → `createProcessQueue(mq, filterExpression)` → `ProcessQueueImpl#wrapReceiveMessageRequest`. The assigned queue's topic name is now only echoed back to the server. - **Back off on receive failure.** The worker waits 20 ms after `TooManyRequests` and 1 s after any other failure. The wait is interrupted by shutdown. This mirrors `ProcessQueueImpl#onReceiveMessageException` (`RECEIVING_FLOW_CONTROL_BACKOFF_DELAY` / `RECEIVING_FAILURE_BACKOFF_DELAY`). Scope notes for reviewers: 1. `LitePushConsumer` wraps `PushConsumer` and uses the same `MessageQueueActor`, so it is fixed by the same change. 2. No public API changes. `PushConsumerOption::get_filter_expression` is kept; it is just no longer used on the receive path. 3. All changes are internal to `rust/src/push_consumer.rs`. This PR does not touch the files changed by #1404 (telemetry/session) or #1381 (LiteSimpleConsumer), apart from a different region of `push_consumer.rs` in #1404. ### How Did You Test This Change? Two unit tests were added. Both failed before the fix and pass after it: - `test_receive_messages_with_prefixed_topic_name`: the assigned queue's topic is `rmq-xxx%test_topic` and the subscription is `test_topic` with tag `tagA`. The test asserts that `ReceiveMessage` is sent, with the queue's topic name unchanged and the filter expression `tagA`. Before the fix the request is never sent. - `test_receive_failure_backoff_delay`: checks 20 ms for `TooManyRequests` and 1 s for other server and transport errors. Existing tests were updated for the new signatures only. ``` $ cargo test test result: ok. 91 passed; 0 failed; 1 ignored (lib) test result: ok. 3 passed; 0 failed (lite_integration_test) test result: ok. 2 passed; 0 failed; 3 ignored (doc-tests) $ cargo fmt --check # clean ``` Note that `cargo clippy --all-features -- -D warnings` currently fails on `master` with 2 `for_kv_map` errors in `rust/src/client.rs` (lines 904 and 912). The same 2 errors remain on this branch, and this PR adds no new clippy warnings (compared with `--lib --tests` before and after the change). The MSRV check could not be run locally because a transitive dependency (`powerfmt 0.2.1`) already requires rustc 1.79 on `master`. End-to-end, against a managed RocketMQ 5 instance that returns `<instance>%<topic>` names, using a `LitePushConsumer`: | | before | after | |---|---|---| | messages consumed | 0 | all, ~50 ms end-to-end | | 30 messages on one lite topic, in order (`fifo=false` and `fifo=true`) | 0 | 30, in order | | redelivery after `ConsumeResult::FAILURE` | none | redelivered | | `OffsetOption` Min / Max / offset | not consumed | as expected | | `no filter expression presents` errors | 8.94M in 8 min | 0 | | client log size for the run | 1.5 GB | 25 KB | The Java client 5.2.2 consumes normally against the same instance, group and topic, which confirms that the difference is on the client side. -- 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]
