leizhiyuan opened a new pull request, #1435:
URL: https://github.com/apache/rocketmq-clients/pull/1435
### Which Issue(s) This PR Fixes
Fixes #1434
> [!IMPORTANT]
> **This PR depends on #1381 (LiteSimpleConsumer). Please merge #1381
first.**
> This branch contains one temporary merge commit, `Merge #1381 (temporary
base ...)`. It reproduces the merged state of #1381, which adds
`LiteSimpleConsumer` and the lite topic in `ChangeInvisibleDurationRequest`.
The only commit to review is the last one, `[ISSUE #1434] ...`. I will rebase
onto `master` and drop the merge commit once #1381 is in.
>
> To view only this change: `git diff HEAD~1 HEAD`, or the last commit in
the Commits tab.
### Brief Description
**Scenario.** A lite simple consumer receives a message of one lite topic
and can not process it yet, so it calls `change_invisible_duration` to retry
later. The Java client sends `suspend = true` for a lite message, and the
broker then suspends the whole lite topic queue of the group until the new
visible time without counting a delivery attempt. The Rust client never sets
`suspend`, so the broker only postpones this single offset and counts the
attempt. Details in #1434.
**Change.** Align with Java `ConsumerImpl#wrapChangeInvisibleDuration`
(#1206):
- `Client::change_invisible_duration_inner` sets `suspend = Some(true)` when
the client is a `LiteSimpleConsumer` and the message has a lite topic. In every
other case `suspend` stays unset, as before.
- The doc comment of `LiteSimpleConsumer::change_invisible_duration`
describes the effect.
#### Notes for reviewers
- **Please confirm the scope (`LitePushConsumer` is excluded on purpose).**
Java also sends `suspend = true` when a `LitePushConsumer` nacks a failed
message on the standard (non-FIFO) path. I first applied the same rule to the
Rust push consumer worker and tested it on a real server with local `fifo =
false`. The listener failed while `delivery_attempt <= 2`:
| | `master` (no `suspend`) | `suspend = true` on the push nack |
|---|---|---|
| attempts of the failing message | 1, 1, 3, then success | stuck at 2,
redelivered every ~1.1 s (114 times in 125 s) |
| later messages of the same lite topic | consumed 3 times in between |
redelivered with it, 339 deliveries in total |
| reaches `max_attempts` / DLQ | yes | never |
Because a suspended change does not count as an attempt, a message that
keeps failing would loop forever instead of going to the DLQ. On this server
Java takes the FIFO path (the server pushes `fifo = true` for lite), which
retries locally and only nacks for `ConsumeResultSuspend`, so it does not reach
this code there. I kept the push consumer unchanged; it can follow once the
expected semantics are confirmed.
- **Gate.** In #1381, `Client::is_lite_consumer()` only matches
`LitePushConsumer`, so I check `ClientType::LiteSimpleConsumer` directly
instead of changing that helper, which is used for other lite push behavior.
- **Conflicts.** The fix also applies cleanly on #1423, which carries #1404
+ #1381; 110 lib tests passed there. Apart from that, `git merge-tree` against
the open PRs gives the same conflicts as #1381 itself: #1404 / #1425
(`Cargo.toml`, `client.rs`, `simple_consumer.rs`), #1409, #1427 and #1433
(`simple_consumer.rs`). This PR adds no new conflicts. It is clean against
#1407, #1411, #1414, #1416, #1420, #1421, #1429 and #1431. Overview in #1412.
### How Did You Test This Change?
**Unit tests** (in `client.rs`, capturing the request sent through a mocked
`RPCClient`):
- `client_change_invisible_duration_of_lite_message_suspends`:
`LiteSimpleConsumer` + lite topic sends `lite_topic` and `suspend =
Some(true)`. This test fails without the fix.
- `client_change_invisible_duration_of_normal_message_does_not_suspend`:
`SimpleConsumer` without a lite topic, and `LiteSimpleConsumer` with a message
that has no lite topic, both leave `suspend` unset.
```
$ cargo test -- --test-threads=1
test result: ok. 99 passed; 0 failed; 1 ignored (lib; #1381 alone: 97)
test result: ok. 3 / 4 / 4 passed (integration tests)
test result: ok. 2 passed; 0 failed; 3 ignored (doc-tests)
$ cargo fmt --check # clean
$ cargo clippy --all-targets # same warnings as
#1381 alone
$ 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.** Not possible for `LiteSimpleConsumer`: the managed RocketMQ
5 server I use rejects the `LITE_SIMPLE_CONSUMER` client type (`Client type is
not consumer, type: LITE_SIMPLE_CONSUMER`), for Java 5.2.2 too. The table above
comes from the real-server test of the excluded push consumer variant.
--
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]
