leizhiyuan opened a new issue, #1432: URL: https://github.com/apache/rocketmq-clients/issues/1432
### 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 When `SimpleConsumer::start()` fails, the consumer is left half started, and `receive_with` / `ack` / `change_invisible_duration` keep working against the server instead of failing. 1. `Client::start` sets `shutdown_tx` (which is what `is_started()` checks) **before** creating the first session, and does not reset it when that fails. So after a failed start the client counts as started. 2. `SimpleConsumer::start` returns the error of the startup route lookup without shutting the client down. The heartbeat / route sync task started by `Client::start` keeps running. 3. `SimpleConsumer::receive_with`, `ack` and `change_invisible_duration` never check whether the consumer is started. Only the very first call before any `start()` fails, and only by accident (`get_session` checks it). **Scenario.** An application retries `start()` or carries on after logging the error (e.g. a wrong group, missing permission, or a topic that does not exist yet). It then calls `receive` in a loop. Instead of a clear "not running" error, every call reconnects and fails with a server error, or, when only the route lookup failed, long polls for 20 s and returns `Ok([])`, as if the consumer were healthy. Observed on a real server with a normal topic: | case | step | Rust today | Java 5.2.2 | |---|---|---|---| | group without permission | `start()` | `Err` (telemetry `PermissionDenied`) | fails | | | then `receive_with(..)` | reconnects, `Err(PermissionDenied)` from the server | `IllegalStateException: Simple consumer is not running now`, no RPC | | | then `ack(&view)` | **panic** in `get_session_with_endpoints(..).unwrap()` | same `IllegalStateException` | | topic that does not exist | `start()` | `Err` (route lookup) | fails | | | then `receive_with(..)` | long polls, `Ok([])` after 20 s | `IllegalStateException` | | | then `ack(&view)` / `change_invisible_duration(&view, ..)` | sent to the server (`40013`) | `IllegalStateException` | **Java behavior.** - `SimpleConsumerImpl#startUp` (like `ProducerImpl` and `PushConsumerImpl`) catches any startup failure, calls `shutDown()` and rethrows. The Guava service ends in state `FAILED`. - `SimpleConsumerImpl#receive0`, `#ack0` and `#changeInvisibleDuration0` first check `isRunning()` and fail with `IllegalStateException("Simple consumer is not running now")`. ### Steps to Reproduce ```rust let mut consumer = SimpleConsumer::new(option_with_topic_that_does_not_exist, client_option)?; assert!(consumer.start().await.is_err()); // Long polls for 20 s and returns Ok([]): let messages = consumer.receive("topic", &FilterExpression::new(FilterType::Tag, "*")).await; ``` ### What Did You Expect to See? The same as Java: a failed `start()` leaves the consumer not running and releases what it started, and `receive_with` / `ack` / `change_invisible_duration` fail immediately with `ErrorKind::ClientIsNotRunning`. ### What Did You See Instead? The consumer behaves as started: it reconnects, long polls, sends acks, and in one case panics. ### Additional Context Found while comparing the Rust and Java SimpleConsumer end to end on the same server. --- Tracked in #1412. -- 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]
