leizhiyuan opened a new pull request, #1416:
URL: https://github.com/apache/rocketmq-clients/pull/1416
### Which Issue(s) This PR Fixes
Fixes #1415
### Brief Description
> [!WARNING]
> **Breaking change.** After this PR, `Producer::batch_send` with more than
one message fails, for lite **and** normal messages, on servers whose proxy
returns one receipt per batch (see below). Workaround: send messages one by one
with `Producer::send`.
**Scenario.** A producer sends 3 lite messages with `batch_send`, and the
call returns `Ok` with one receipt. That receipt's `message_id` is the 3 ids
joined with `,`. A lite consumer subscribed with offset `MIN` never receives
any of them. When the 3 messages belong to different lite topics, the client
accepts the batch and the server still answers `OK`.
**Root cause.** The client builds the request the same way Java does: one
`SendMessageRequest` with 3 `pb::Message`s, each with its own `lite_topic`,
verified at the gRPC layer. The problem is on the server side, and the Rust
client hides it.
- **Proxy.** `SendMessageActivity` sends a multi-message request as one
batch to the broker and answers with one `SendResultEntry` per broker result.
For a batch that is a single entry (`// factually, the size of result is always
1`), whose `message_id` is the joined ids. This applies to every batch, lite or
not. It is the case in Apache RocketMQ proxy 5.1.2 and in the managed server I
tested.
- **Broker.** The single-message path of `SendMessageProcessor` turns the
lite topic property into `INNER_MULTI_DISPATCH`, which dispatches the message
to the lite topic's queue. **`sendBatchMessage` has no such step.** So a
batched lite message only lands in the parent topic, and lite consumers never
see it.
- **Clients.** The Java client fails such a send: `ProducerImpl#send0`
checks that the receipt count equals the message count, and throws `received
send receipt's quantity 1 is not equal to sent message's quantity 3`. The Rust
client skips this check and returns the server entries as they are.
**Changes** (`rust/src/producer.rs` only):
1. **`batch_send` rejects lite messages in a multi-message batch before any
route lookup or RPC.** The error is `ErrorKind::InvalidMessage`: `lite messages
do not support batch sending, send them one by one`. This covers same lite
topic, different lite topics, and lite mixed with normal messages. Sending a
single lite message through `send` is unchanged.
2. **`batch_send` checks the receipt count, aligned with Java
`ProducerImpl#send0`.** If the server returns a different number of receipts
from the number of messages sent, it fails with `ErrorKind::Server`: `received
{actual} send receipt(s) for {sent} sent message(s)`. It no longer returns the
joined receipt. The two errors are distinct: `InvalidMessage` is a client-side
rejection, nothing was sent; `Server` means the server answered unexpectedly.
3. **`send` no longer panics when the server returns no entry.** It used to
do `results[0].clone()`. Now it goes through the same check and gets `received
0 send receipt(s) for 1 sent message(s)`.
4. **The doc comment of `batch_send` is updated.** It now says that the
server does not guarantee one receipt per message, that a mismatch is an error,
and that callers should send one by one.
#### Breaking change and why
| `batch_send` call | before | after |
|---|---|---|
| 1 message (any type), server returns 1 entry | `Ok`, 1 receipt | `Ok`, 1
receipt (unchanged) |
| N > 1 lite messages | `Ok`, 1 joined receipt, **never delivered** |
`Err(InvalidMessage)`, nothing sent |
| N > 1 normal messages, proxy returns 1 entry | `Ok`, 1 joined receipt,
messages stored | **`Err(Server)`**, messages probably stored |
| `send`, server returns `OK` with 0 entries | **panic** (index out of
bounds) | `Err(Server)` |
The third row is the intended trade-off. For normal messages the batch is
probably stored. But the caller can map neither a receipt to a message nor a
per-message offset. A result of `Ok` with a `,`-joined id breaks any code that
stores, logs or recalls by message id. Failing explicitly is safer than
returning a success that looks like it covers all the messages. It is also what
the Java client does: Java exposes no public batch API, and its internal batch
path has the same check.
Retrying after a `Server` error may duplicate normal messages, since they
were probably stored. The doc comment says so. **Workaround:** use
`Producer::send` for each message.
#### Notes for reviewers
- **No conflict with other open PRs.** Only `rust/src/producer.rs` is
touched, and no open PR touches it. `git merge-tree` against #1381, #1404,
#1407, #1409, #1411 and #1414 is clean, so this can be merged in any order. All
of them are tracked in #1412.
- **Unrelated flaky test.** While running the suite repeatedly,
`client::tests::client_get_session` / `test_get_session_for_lite_consumer`
failed occasionally, with "MockSession::new: No matching expectation found". It
is a pre-existing race between these two tests: `master` without this PR shows
2 failures in 100 runs of `cargo test --lib`. #1411 fixes it by taking `MTX` in
`client_get_session`.
### How Did You Test This Change?
**Unit tests: 5 added.** The three marked "fails on master" were checked by
running them against `master`'s implementation.
- `producer_batch_send_rejects_lite_messages` (fails on master): batches
with the same lite topic, different lite topics, and lite + normal. Each is
rejected with `InvalidMessage` and the message above. No expectation is set on
the mocked client, which proves that no route lookup or RPC happens.
- `producer_batch_send_fails_when_receipt_quantity_mismatches` (fails on
master): 3 normal messages, and the mock returns 1 receipt `"id_0,id_1,id_2"`.
The result must be `Err(Server)` with `received 1 send receipt(s) for 3 sent
message(s)`. On `Ok` the test panics and prints the receipts, which guarantees
the joined receipt is not returned.
- `producer_send_fails_when_receipt_quantity_mismatches` (fails on master,
where it panics): `send` with 0 entries returns `Err(Server)`.
- `producer_batch_send_returns_one_receipt_per_message`: when the counts
match, the receipts are returned in order.
- `producer_send_single_lite_message`: a single lite message is still sent
exactly once, with its `lite_topic`.
```
$ cargo test
test result: ok. 94 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
```
The full `cargo test` was repeated 10 times and `cargo test --lib` 30 times.
The only failures were the pre-existing flaky `client_get_session` race
described above.
`cargo clippy --all-features -- -D warnings` reports only the 2 `for_kv_map`
errors that `master` already has in `rust/src/client.rs` (lines 904 and 912).
`cargo clippy --lib --tests` reports the same 42 findings as `master`, so this
PR adds none.
**End-to-end.** I ran this against a managed RocketMQ 5 server, with a lite
topic, a `LitePushConsumer` subscribed with offset `MIN`, and gRPC
request/response logging:
| | before | after |
|---|---|---|
| `batch_send` 3 lite messages, same lite topic | `Ok`, 1 joined receipt;
consumed: none | `Err(InvalidMessage)`, no request sent |
| `batch_send` 3 lite messages, different lite topics | `Ok`, 1 joined
receipt; consumed: none | `Err(InvalidMessage)`, no request sent |
| `send` 1 lite message (control) | `Ok`; consumed | `Ok`; consumed |
The Java comparison is in #1415: the same lite batch through Java's internal
`ProducerImpl#send(List<Message>, boolean)` fails with the receipt quantity
error and is not consumed either.
The server I used only has a lite topic, so the normal-message batch case is
covered by the unit test plus the proxy source above, not end-to-end.
---
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]
