leizhiyuan opened a new issue, #1408: URL: https://github.com/apache/rocketmq-clients/issues/1408
### 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 `LitePushConsumer` never applies the lite settings pushed by the server, and never handles `NotifyUnsubscribeLiteCommand`. As a result, the client-side lite subscription quota check is effectively disabled. **Scenario.** The server pushes `Settings.subscription.lite_subscription_quota = 2000` and `max_lite_topic_size = 64` to the client. An application keeps calling `subscribe_lite` for new lite topics (one lite topic per session/user/device is a typical lite use case). With the Java client, the 2001st `subscribeLite` is rejected locally with `LiteSubscriptionQuotaExceededException` (code `42902`). With the Rust client, all 2010 subscriptions "succeed" locally, and the local `lite_topic_set` grows beyond the quota the server granted. Likewise, when the server notifies the client to drop a lite topic, the Rust client keeps it in `lite_topic_set` and re-adds it on every 30 s `CompleteAdd` sync. There are two root causes. **1. Telemetry commands never reach the lite handler.** `LitePushConsumer::start` creates a channel, passes the sender to `PushConsumer::start_with_telemetry`, and spawns a handler that calls `LiteSubscriptionManager::sync_settings` / `on_notify_unsubscribe_lite_command`. But `start_with_telemetry` ignores the argument (`_telemetry_command_tx`) and creates its own channel: ```rust // rust/src/push_consumer.rs pub async fn start_with_telemetry( &mut self, _telemetry_command_tx: mpsc::Sender<pb::TelemetryCommand>, ) -> Result<(), ClientError> { let (command_tx, mut command_rx) = mpsc::channel(16); self.client.start(command_tx).await?; ``` The sender is dropped right away, so the lite handler sees a closed channel and exits. In the log, `The LitePushConsumer starts successfully` and `LitePushConsumer telemetry handler stopped` appear in the same millisecond, and `Updated lite subscription quota` never appears. The quota stays at its initial value `0`. **2. Startup does not wait for the server settings, and quota `0` means "unlimited".** ```rust // rust/src/lite_subscription_manager.rs // If quota is 0, it means the server hasn't returned the quota yet. // In this case, we should allow subscriptions to proceed. if quota == 0 { return Ok(()); } ``` So even after (1) is fixed, `subscribe_lite` calls made right after `start()` returns are not checked until the first `Settings` arrives. Also, `LiteSubscriptionManager::start` (the first `CompleteAdd` sync) runs before the lite handler is set up. **Java behavior that Rust should align with:** - `LitePushConsumerImpl#onSettingsCommand` calls `super.onSettingsCommand(...)` and then `liteSubscriptionManager.sync(settings)`. `LitePushConsumerImpl#onNotifyUnsubscribeLiteCommand` calls `liteSubscriptionManager.onNotifyUnsubscribeLiteCommand(...)`. - `ClientImpl#startUp` fetches routes through `onTopicRouteDataFetched`, which waits on `ClientSessionImpl#syncSettings()`. That future is completed by the first `Settings` command, and it times out after `SETTINGS_INITIALIZATION_TIMEOUT` (3 s) plus the request timeout, which fails startup. `LitePushConsumerImpl#startUp` runs `liteSubscriptionManager.startUp()` only after `super.startUp()`. So when `start()` returns, the quota is already known. - `LiteSubscriptionManager#checkLiteSubscriptionQuota` is simply `liteTopicSet.size() + delta > liteSubscriptionQuota`, with no special case for `0`. ### Steps to Reproduce 1. Use a RocketMQ 5 server that pushes `lite_subscription_quota` in the `Settings` telemetry command (e.g. quota `2000`). 2. Start a `LitePushConsumer` bound to a lite-enabled topic. 3. Call `subscribe_lite` for 2010 distinct lite topics. ### What Did You Expect to See? As with the Java client: - the 2001st `subscribe_lite` fails locally with code `42902` (`LITE_SUBSCRIPTION_QUOTA_EXCEEDED`); - `max_lite_topic_size` from the server is applied; - a `NotifyUnsubscribeLiteCommand` removes the lite topic from the local set; - `start()` returns only after the server settings are applied. ### What Did You See Instead? - All 2010 `subscribe_lite` calls succeed, and `get_lite_topic_set().len() == 2010`. - `Updated lite subscription quota` / `Updated max lite topic size` are never logged. - The lite telemetry handler stops in the same millisecond as startup completes. ### Additional Context The Java client 5.2.2 rejects the 2001st subscription 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]
