leizhiyuan opened a new pull request, #1441:
URL: https://github.com/apache/rocketmq-clients/pull/1441
### Which Issue(s) This PR Fixes
Fixes #1440
### Brief Description
**Scenario.** Several application instances each run a `SimpleConsumer` and
call `receive` in a loop. Today every instance polls route entry 0 first, then
1, and so on. Unreadable queues and queues of slave brokers get receive
requests too, so some calls come back empty or fail. Details in #1440.
**Root causes.**
1. `SimpleConsumer::receive_with` uses `select_message_queue`, a round-robin
over all queues of the route.
2. `TopicRouteManager::query_topic_route` creates every `Route` with `index:
0`.
**Change.** Align with Java `SubscriptionLoadBalancer`:
- New `util::select_readable_message_queue`: round-robin over the queues
that are readable (`READ`, `READ_WRITE`) and on a master broker (`broker.id ==
0`). It fails with `ErrorKind::NoBrokerAvailable` ("no readable message queue
found", operation `simple_consumer.receive_message`) when there is none,
without sending a request. `SimpleConsumer::receive_with` uses it.
- `is_readable_master_queue` is copied byte for byte from #1381, which adds
the same check for `LiteSimpleConsumer` (`prune_lite_route`). Like #1381, it
tolerates `PERMISSION_UNSPECIFIED`, since some route responses omit the field.
Java throws there.
- New routes start round-robin at a random index in `[0, i32::MAX)`, like
`RandomUtils.nextInt(0, Integer.MAX_VALUE)` in Java `SubscriptionLoadBalancer`
/ `PublishingLoadBalancer`. The randomness comes from the std `RandomState`, so
no new dependency. When a route is refreshed and its queues are unchanged, the
existing `Route` and its index are kept, as before.
#### Notes for reviewers
- **Scope of the random start index.** `query_topic_route` is shared, so the
producer's `select_message_queue` (non-FIFO send) and `Client::pick_endpoints`
also start at a random queue now. That is what Java `PublishingLoadBalancer`
does. FIFO sends (`select_message_queue_by_message_group`) hash the group and
are unaffected. The producer still does not filter writable master queues; I
left that out to keep this PR small.
- **Conflicts.** It conflicts with #1381, and with #1423 / #1435, which
carry #1381:
- `simple_consumer.rs`: #1381 moves the body of `receive_with` into
`receive_internal`.
- `util.rs`: both add code at the same place.
Resolution: keep #1381's `receive_internal`, replace its
`select_message_queue(route)` with `select_readable_message_queue(route,
Self::OPERATION_RECEIVE_MESSAGE)?`, and keep both `util.rs` sides. The
`is_readable_master_queue` function then exists once, because both sides add
the same text. Results on temporary merges: #1381 100 passed, #1423 111, #1435
102.
`git merge-tree` is clean against every other open Rust PR. Temporary
merges with the PRs that touch `client.rs`, `simple_consumer.rs` or `util.rs`
also pass: #1404 101, #1409 98, #1416 97, #1425 103, #1427 95, #1429 100, #1431
94, #1433 95, #1437 94. Overview in #1412.
### How Did You Test This Change?
**Unit tests.** All three fail on `master`:
-
`util::tests::util_select_readable_message_queue_skips_unreadable_and_slave_queues`:
a route with a slave queue, a write-only queue, a `NONE` queue and two
readable master queues. Only the two readable master queues are selected,
round-robin.
-
`util::tests::util_select_readable_message_queue_fails_without_readable_queue`:
returns `NoBrokerAvailable`.
- `client::tests::client_query_route_starts_at_random_index`: 8 queried
routes do not all start at the same index (on `master` all are 0).
`client_update_same_route` used to assert that the index of the kept route
is exactly `1`. It now asserts that the index is kept, whatever its start value.
```
$ 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
$ cargo clippy --all-targets # same warnings as
master
$ cargo clippy --all-features -- -D warnings # only the 2
existing for_kv_map errors of master
$ RUSTDOCFLAGS='-D warnings' cargo doc --no-deps # no warnings
$ cargo build --examples --all-features # ok
```
**End-to-end.** Against a managed RocketMQ 5 server, with a normal topic and
group. The route has 18 queues, all readable and on master brokers. Four fresh
`SimpleConsumer`s each called `receive_with` four times, with the selected
queue logged (E2E-only log line):
| | queue ids picked | first queue per consumer | receive errors |
|---|---|---|---|
| `master` | `0 1 2 0` ×4 | 0, 0, 0, 0 | 0 |
| this PR | `2 0 1 2`, `2 0 1 2`, `1 2 0 1`, `2 0 1 2` | different start
positions | 0 |
The filtering path needs a route with unreadable or slave queues, which this
server does not expose, so it is covered by the unit tests only.
--
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]
