leizhiyuan opened a new pull request, #1423:
URL: https://github.com/apache/rocketmq-clients/pull/1423
### Which Issue(s) This PR Fixes
Fixes #1422
> [!IMPORTANT]
> **This PR depends on #1404 and #1381. Please merge those two first.**
> This branch contains two temporary merge commits, `Merge #1404 (temporary
base ...)` and `Merge #1381 (temporary base ...)`. They reproduce the merged
state of both PRs, because they rewrite the same consumer and session code this
fix runs on. The only commit to review is the last one, `[ISSUE #1422] ...`,
which touches `rust/src/push_consumer.rs` only. I will rebase onto `master` and
drop the merge commits once #1404 and #1381 are in.
>
> To view only this change: `git diff HEAD~1 HEAD`, or the last commit in
the Commits tab.
### Brief Description
**Scenario.** A `LitePushConsumer` assigned 6 queues takes 208.7 s to
`shutdown()` after sitting idle (`master` + #1404 + #1381). On `master` it is
20 to 345 s. The Java client takes about 6 s against the same server, group and
topic. Details in #1422.
**Root cause.**
1. Each `MessageQueueActor` worker waits on `ticker.tick()` (1 ms) and
`shutdown_token.cancelled()` in an unbiased `select!`. When a long polling
request returns, both are ready, so the worker often starts another long poll
instead of stopping.
2. Queue actors are shut down one by one in `start()`,
`start_with_telemetry()` and `process_assignments`, so each actor's in-flight
long poll adds to the total.
**Change.** This aligns with Java:
- **Stop receiving once shutdown starts.** The worker `select!` is `biased`
and checks the shutdown token first, so no new receive request is issued after
shutdown starts. This matches `ProcessQueueImpl#receiveMessage`, which checks
`consumer.isRunning()` before every receive.
- **Shut actors down concurrently.** A new `PushConsumer::shutdown_actors()`
shuts all given actors down at the same time through a `TaskTracker` and logs
actor shutdown errors instead of discarding them. All three serial loops use
it. Shutdown now takes about one long polling timeout in total, like
`PushConsumerImpl#waitingReceiveRequestFinished`.
An in-flight long poll is still allowed to finish rather than being aborted,
as in Java. Dropping the stream would leave the messages the server has already
popped for it invisible until their invisible duration expires.
#### Notes for reviewers
- **Remaining slow case, tracked separately.** With this PR the end-to-end
shutdown takes about 5 s in most runs, but sometimes about 28 s. The cause is a
separate issue in #1404's `apply_settings`. It updates the
`long_polling_timeout` sent in the request (20 s from the server here). It does
not update the shared value that the client-side gRPC deadline is based on, so
the deadline stays at the configured 40 s + 3 s. When the server does not
answer at 20 s, the request runs until 43 s. I will file that as its own issue
and PR, which also depends on #1404.
- **Conflicts.**
- With #1407 (filter expression and receive backoff), there is a text
conflict in the worker loop. The resolution is to keep #1407's error handling
and backoff inside the `ticker` branch, and keep this PR's `biased;` with the
shutdown branch first. The new tests then need the extra `filter_expression`
argument (`FilterExpression::sub_all()`). I checked this resolution with the
unit tests and with the end-to-end run below.
- With #1409 there is the same `push_consumer.rs` overlap that #1409
already has with #1404.
- Overview and merge order in #1412.
### How Did You Test This Change?
**Unit tests.** Two tests are added. They use a mocked session whose
`receive_message` blocks like a long poll, and both fail on the base branch
(`master` + #1404 + #1381):
- `test_actor_stops_receiving_once_shutdown_starts`: 8 workers are polling.
Once shutdown starts, no further receive request may be issued. On the base
branch, 6 to 9 more requests are issued after shutdown starts.
- `test_shutdown_actors_waits_for_long_polling_concurrently`: shutting down
6 actors that are each in a 200 ms poll must take less than 3 polls. On the
base branch it took 1.0 to 1.6 s, i.e. 5 to 8 polls.
Both passed 20 out of 20 repeated runs.
```
$ cargo test
test result: ok. 108 passed; 0 failed; 1 ignored (lib; base: 106)
... all other test targets pass
$ cargo fmt --check # clean
$ cargo clippy -- -D warnings # clean
$ RUSTDOCFLAGS="--cfg docsrs" cargo doc --no-deps --all-features # no
warnings
$ cargo build --examples # ok
```
`cargo clippy --all-targets` reports the same findings per file as the base
branch.
Some tests in `client::tests` fail intermittently on the base branch too:
`client_get_session`, `test_lite_consumer_session_sharing` and
`test_reconnect_session_rebuilds_missing_session`. Running only `client::` 100
times gave 14 failures on the base and 13 with this PR. The cause is that some
of these tests set a `MockSession::new_context()` expectation without holding
`MTX`. They are unrelated to this change.
**End-to-end.** I ran this against a managed RocketMQ 5 server, with #1407
also merged so that the consumer can receive on that server. A
`LitePushConsumer` with one lite subscription and 6 assigned queues sits idle
for 35 s, then calls `shutdown()`:
| | shutdown time | background reconnects in the 70 s after shutdown |
|---|---|---|
| base (`master` + #1404 + #1381 + #1407) | 208.7 s | none |
| this PR, 6 runs | 7.96 s, 5.01 s, 5.05 s, 4.98 s, 28.3 s, 27.9 s | none |
The end-to-end runs used #1404 at `5fad37c3`. #1404 has since added two
commits that only change session reconnection in `client.rs`, and this branch
is based on the new head (`b5cb4940`). The unit tests, clippy and the examples
were re-run on it.
The two ~28 s runs are the separate deadline issue described above. Their
logs show the receive requests ending with `Timeout expired` after 43 s.
---
Tracked in #1412 (overview of the related Rust issues / PRs and a suggested
merge order).
--
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]
