leizhiyuan opened a new issue, #1451: URL: https://github.com/apache/rocketmq-clients/issues/1451
### 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 With more than one subscribed topic, the assignment scan restarts the queues of every topic but the last one on every round (every 30 s). The scan calls `process_assignments` once per topic with the same `actor_table` ([push_consumer.rs#L241](https://github.com/apache/rocketmq-clients/blob/e90ece7a/rust/src/push_consumer.rs#L241), same in `start_with_telemetry` [#L370](https://github.com/apache/rocketmq-clients/blob/e90ece7a/rust/src/push_consumer.rs#L370)), and `process_assignments` shuts down every actor that is not in **this topic's** assignments ([#L480-L488](https://github.com/apache/rocketmq-clients/blob/e90ece7a/rust/src/push_consumer.rs#L480-L488)). Processing topic B shuts down all actors of topic A; the next round's A step shuts down all actors of B. Each restart also waits for the long polling of the stopped actors. Java only considers the queues of the topic being synchronized (`PushConsumerImpl#syncProcessQueue`): ```java if (!topic.equals(mq.getTopic())) { continue; } ``` ### Steps to Reproduce Push consumer subscribed to topics `a` and `b`, one queue each with messages, run for 75 s (more than two scans). ### What Did You Expect to See? Both topics consumed. ### What Did You See Instead? | messages consumed a / b in 75 s | | |---|---| | `master` (3 runs) | 0 / 373,790; 355,193 / 0; 192 / 350,091 | | fix | 609,013 / 608,948 | Which topic survives depends on the `HashMap` iteration order. ### 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]
