leizhiyuan opened a new issue, #1406: URL: https://github.com/apache/rocketmq-clients/issues/1406
### 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` 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` / `LitePushConsumer` never receives any message when the server returns the assigned message queue's topic name in a form that differs from the subscription key, e.g. with an instance prefix: ``` MessageQueue { topic: Resource { resource_namespace: "", name: "rmq-xxxxxxxx%my_topic" }, ... } ``` The subscription table is keyed by `my_topic`, but `PushConsumer::receive_messages` looks up the filter expression by the assigned queue's topic name: ```rust // rust/src/push_consumer.rs let filter_expression = option .get_filter_expression(&message_queue.topic.name) .ok_or(ClientError::new(ErrorKind::Unknown, "no filter expression presents", ...))?; ``` The lookup always fails, so no `ReceiveMessage` request is ever sent. It is made much worse by the receive loop in `MessageQueueActor::start`: the worker is driven by a 1 ms ticker and only logs the error, with no backoff. Every worker (6 queues x 4 workers in our case) spins at ~1 kHz. In 8 minutes we got 8.94 million `ERROR receive messages error ... no filter expression presents` lines (~1.5 GB of logs) and zero consumed messages. The Java client is not affected because it never derives the filter expression from the assigned queue. `PushConsumerImpl#scanAssignments` iterates `subscriptionExpressions`, passes the `FilterExpression` to `syncProcessQueue` / `createProcessQueue`, and `ProcessQueueImpl` keeps it and uses it in `wrapReceiveMessageRequest`. On receive failure, `ProcessQueueImpl#onReceiveMessageException` also backs off (20 ms for `TooManyRequests`, 1 s otherwise). ### Steps to Reproduce 1. Use a RocketMQ 5 proxy that returns topic names with a prefix in `QueryAssignmentResponse` (observed on TDMQ for RocketMQ 5). 2. Start a `PushConsumer` subscribing `my_topic` with `FilterExpression::new(FilterType::Tag, "*")` (or a `LitePushConsumer` bound to `my_topic`, then `subscribe_lite(...)`). 3. Produce messages to `my_topic`. ### What Did You Expect to See? Messages are consumed, using the filter expression from the subscription. On receive failures, the worker backs off instead of retrying every millisecond. ### What Did You See Instead? No message is consumed, and the log is flooded with: ``` ERROR rocketmq::push_consumer: receive messages error: Some(Unknown error at push_consumer.receive_message => no filter expression presents ``` ### Additional Context The same Java client (5.2.2) consumes normally against the same instance, group and topic. -- 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]
