leizhiyuan opened a new pull request, #1409:
URL: https://github.com/apache/rocketmq-clients/pull/1409
### Which Issue(s) This PR Fixes
Fixes #1408
### Brief Description
**Scenario.** The server grants a `LitePushConsumer` a lite subscription
quota (e.g. `lite_subscription_quota = 2000`) and a `max_lite_topic_size`
through the `Settings` telemetry command. It can also revoke a lite topic with
`NotifyUnsubscribeLiteCommand`. Applications that create one lite topic per
session/user/device rely on the client rejecting subscriptions beyond the
quota. The Java client does this locally with code `42902`. The Rust client
never applied any of these server settings:
1. `PushConsumer::start_with_telemetry` ignored the channel passed in by
`LitePushConsumer` (`_telemetry_command_tx`) and created its own. The lite
telemetry handler saw a closed channel and exited right after startup. No
`Settings` or `NotifyUnsubscribeLiteCommand` ever reached
`LiteSubscriptionManager`.
2. The quota therefore stayed at its initial value `0`, and
`check_lite_subscription_quota` treated `0` as "not known yet, allow". The
quota check was effectively disabled. Startup also never waited for the server
settings, so even with (1) fixed, subscriptions made right after `start()`
would bypass the check.
Against a real server pushing quota 2000, all 2010 `subscribe_lite` calls
succeeded. Java rejects the 2001st.
This PR aligns `LitePushConsumer` with the Java client:
| Java | This PR |
|---|---|
| `LitePushConsumerImpl#onSettingsCommand` calls
`super.onSettingsCommand(...)` and then
`liteSubscriptionManager.sync(settings)`. `#onNotifyUnsubscribeLiteCommand`
calls `liteSubscriptionManager.onNotifyUnsubscribeLiteCommand(...)` |
**`start_with_telemetry` forwards every telemetry command** to the caller's
channel after the push consumer has handled it (retry policy), so the existing
lite handler receives `Settings` and `NotifyUnsubscribeLiteCommand` |
| `ClientImpl#startUp` waits on `ClientSessionImpl#settingsInitFuture`,
which is completed by the first `Settings` and times out after
`SETTINGS_INITIALIZATION_TIMEOUT` (3 s) + request timeout, failing the startup
| **`LitePushConsumer::start` waits for the first `Settings`**, bounded by 3 s
+ `ClientOption::timeout()`. On timeout it shuts down and returns an error,
using the existing start-failure path. `LiteSubscriptionManager` gains a
`watch`-based `wait_for_initial_settings`, completed by any `Settings` command,
as in Java |
| `LitePushConsumerImpl#startUp` runs `liteSubscriptionManager.startUp()`
after `super.startUp()` | **`LiteSubscriptionManager::start` (first
`CompleteAdd` sync) now runs after the settings are applied**. Before, it ran
before the lite handler was even spawned |
| `LiteSubscriptionManager#checkLiteSubscriptionQuota` is
`liteTopicSet.size() + delta > liteSubscriptionQuota`, with no special case for
`0` | **Removed the `quota == 0` → allow branch** |
Other notes for reviewers:
1. **`lite_subscription_manager` is now compiled in unit tests.** In
`lib.rs` it was gated by `#[cfg(not(test))]`, so its existing tests
(`test_validate_lite_topic`, `test_check_quota`) never ran.
`lite_push_consumer` stays gated, because it cannot be built against the mocked
`Client`. #1381 makes the same change.
2. **Behavior change.** If the server does not push `Settings` within 3 s +
request timeout, `LitePushConsumer::start` now fails instead of starting with
an unknown quota. This matches Java. In our test, the first `Settings` arrived
0.3 s after startup began.
3. **Overlap with other PRs.**
- #1381 (LiteSimpleConsumer) adds a similar `wait_for_initial_settings`
to `LiteSubscriptionManager`, but only for `LiteSimpleConsumer`, and on timeout
it logs a warning and continues. Whichever lands second will need a small
rebase. I'm happy to adapt.
- #1404 touches the telemetry loop in `push_consumer.rs`, but not
`start_with_telemetry`.
4. **Out of scope.** Server-pushed push-consumer settings (fifo, batch size,
...) are not applied by `super.onSettingsCommand` in Rust. That is handled by
#1404.
### How Did You Test This Change?
**Unit tests.** Each new test fails without the corresponding change:
- `push_consumer::tests::test_start_with_telemetry_forwards_commands`: it
pushes `Settings` (quota 2000, max 64) and `NotifyUnsubscribeLiteCommand` into
the client's command channel, then asserts both arrive on the channel passed to
`start_with_telemetry`. Before the fix, `recv()` returns `None`.
- `lite_subscription_manager::tests::test_check_quota`: with quota `0`,
`delta = 1` is rejected with code `42902`. It fails with the old `quota == 0`
branch.
- `test_wait_for_initial_settings`: times out before `Settings`, returns
immediately after, and quota and max size are updated.
- `test_wait_for_initial_settings_wakes_up_waiter`: a pending waiter is
woken by a `Settings` without a subscription.
- `test_scheduler_clone_shares_initial_settings_state`: the scheduler clone
shares the initialization state and the quota.
```
$ cargo test
test result: ok. 95 passed; 0 failed; 1 ignored (lib; master: 89)
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
$ RUSTDOCFLAGS="--cfg docsrs" cargo doc --no-deps --all-features # no
warnings
$ cargo build --examples # ok
```
`cargo clippy --all-features -- -D warnings` fails on `master` with 2
`for_kv_map` errors in `rust/src/client.rs` (lines 904 and 912). They are
unchanged here, and this PR adds no new clippy findings in library code.
**End-to-end**, against a managed RocketMQ 5 server that pushes
`lite_subscription_quota = 2000` and `max_lite_topic_size = 64`, with a
`LitePushConsumer` subscribing 2010 distinct lite topics:
| | before | after |
|---|---|---|
| `subscribe_lite` result | all 2010 succeed | 2000 succeed; the 2001st
fails with `Lite subscription quota exceeded: current=2000, delta=1,
quota=2000` (code `42902`) |
| `Updated lite subscription quota to 2000` | never logged | logged before
`starts successfully`, and again on every Settings refresh |
| lite telemetry handler | stops in the same millisecond as startup | runs
until `shutdown()` |
The Java client 5.2.2 behaves the same way as the "after" column against the
same instance, group and topic.
The e2e run was done on top of #1407. Without it, `LitePushConsumer` cannot
receive messages from this server at all. The changes in this PR are
independent of #1407.
--
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]
