leizhiyuan opened a new pull request, #1433:
URL: https://github.com/apache/rocketmq-clients/pull/1433
### Which Issue(s) This PR Fixes
Fixes #1432
### Brief Description
**Scenario.** An application keeps going after `SimpleConsumer::start()`
failed (wrong group, missing permission, topic not created yet) and calls
`receive` in a loop. Today the consumer behaves as if it were started: it
reconnects and fails with server errors, or long polls for 20 s and returns
`Ok([])`; an `ack` can even panic. The application never sees a "not running"
error.
**Root causes.**
1. `Client::start` sets `shutdown_tx`, which `is_started()` checks, before
creating the first session, and leaves it set when that fails.
2. `SimpleConsumer::start` returns the error of the startup route lookup
without shutting the client down, so the heartbeat / route sync task of
`Client::start` keeps running.
3. `SimpleConsumer::receive_with` / `ack` / `change_invisible_duration` do
not check the state.
**Change.** Align with Java (`SimpleConsumerImpl#startUp` calls `shutDown()`
on failure; `#receive0`, `#ack0`, `#changeInvisibleDuration0` check
`isRunning()`):
- `Client::start` resets `shutdown_tx` and `telemetry_command_tx` when the
first session can not be created. No task has been spawned yet at that point.
- `SimpleConsumer::start` calls `Client::shutdown_ref()` when the route
lookup of a configured topic fails, then returns the original error. A shutdown
error is only logged.
- `receive_with` (and therefore `receive`), `ack` and
`change_invisible_duration` call `check_started` first and fail with
`ErrorKind::ClientIsNotRunning` before any RPC. Operations
`simple_consumer.receive_message`, `simple_consumer.ack_message`,
`simple_consumer.change_invisible_duration`.
#### Notes for reviewers
- **Scope.** The `Client::start` rollback benefits every client type, since
they all call it first. `Producer` and `PushConsumer` have the same "no
shutdown when a later startup step fails" pattern; I left them out to keep this
PR small and will follow up separately if wanted. The operation checks are only
added to `SimpleConsumer`; `PushConsumer` has no user-facing receive / ack.
- **Behavior change.** Calling `receive` / `ack` /
`change_invisible_duration` on a consumer that was never started, or whose
start failed, now returns `ClientIsNotRunning` instead of trying to reach the
server. The first call on a never started consumer already failed with the same
kind (via `get_session`); acks on such a consumer used to panic.
- **Conflicts.** `git merge-tree` is clean against #1404, #1407, #1409,
#1411, #1414, #1416, #1420, #1421, #1425 and #1429. I also built and tested the
merges with the PRs that touch `client.rs`: #1409 (98 passed), #1411 (94),
#1429 (100). With #1404 (and #1425, which carries it) the new test needs the
extra `Session::new` argument and a `set_stream_closed_tx` expectation that
#1404 adds; with that, 101 passed.
- It conflicts in `simple_consumer.rs` with #1381 (and #1423, which
carries it), #1427 and #1431. All three touch the same methods or append tests
at the same place. Resolution: put the `check_started` call first (at the top
of `receive_internal` for #1381, before #1427's argument checks), keep the rest
of both sides and all tests, and add a `check_started` expectation to their
mocked clients. Verified on temporary merges: 100 passed (#1381), 95 (#1427),
94 (#1431). Overview in #1412.
### How Did You Test This Change?
**Unit tests.** All three fail on `master`:
- `client::tests::client_start_failure_leaves_client_not_started`:
`Session::start` fails; afterwards `is_started()` is false and `get_session()`
returns `ClientIsNotRunning`.
- `simple_consumer::tests::operations_fail_when_not_started`: `receive`,
`ack` and `change_invisible_duration` return `ClientIsNotRunning` with their
operation name; the mocked client expects no route lookup or RPC.
- `simple_consumer::tests::start_failure_shuts_down_client`: the route
lookup fails during `start()`; the client is shut down once.
```
$ cargo test -- --test-threads=1
test result: ok. 92 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='-D warnings' cargo doc --no-deps # no warnings
$ cargo build --examples --all-features # ok
$ cargo clippy --all-features -- -D warnings # only the 2
existing for_kv_map errors of master
```
`cargo clippy --all-targets` reports two more `await_holding_lock` warnings,
both in the new tests. They come from the `MTX` guard that serializes the
static mock contexts, the same pattern and warning as the existing
`simple_consumer_start`, producer and push consumer tests. `--test-threads=1`
avoids the existing flaky `client::tests::client_get_session` (it uses
`MockSession::new_context()` without holding `MTX`), unrelated to this change.
**End-to-end.** Against a managed RocketMQ 5 server, with a normal topic;
the same program ran on `master` and on this branch:
| case | step | `master` | this PR |
|---|---|---|---|
| group without permission (telemetry fails) | `start()` | `Err` | `Err` |
| | `receive_with(..)` | reconnects, server `PermissionDenied` |
`Err(ClientIsNotRunning)` in 78 µs |
| | `ack` / `change_invisible_duration` | **panic** (`unwrap()` in
`get_session_with_endpoints`) | `Err(ClientIsNotRunning)` |
| topic that does not exist (route lookup fails) | `start()` | `Err` |
`Err`, client shut down |
| | `receive_with(..)` | `Ok([])` after a 20 s long poll |
`Err(ClientIsNotRunning)` in 19 µs |
| | `ack` / `change_invisible_duration` | sent to the server (`40013`) |
`Err(ClientIsNotRunning)` |
Java 5.2.2 fails all follow-up calls with `IllegalStateException: Simple
consumer is not running now`.
---
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]
