leizhiyuan opened a new pull request, #1431:
URL: https://github.com/apache/rocketmq-clients/pull/1431
### Which Issue(s) This PR Fixes
Fixes #1430
### Brief Description
**Scenario.** For slow processing, a simple consumer receives a message,
extends its invisible duration with `change_invisible_duration` while the work
is still running, and acks the same `MessageView` when done. The server
invalidates the old receipt handle when it issues a new one. The Rust
`MessageView` keeps the old handle, so the final `ack` returns `Ok(())` but has
no effect, and the message is redelivered. A second `change_invisible_duration`
of the same view uses the old handle as well. The new handle is only returned
as a `String`, and the public API can not ack with it.
**Change.** Align with Java `SimpleConsumerImpl#changeInvisibleDuration0`,
which calls `MessageViewImpl#setReceiptHandle(response.getReceiptHandle())`:
- `MessageView::receipt_handle` becomes a
`parking_lot::Mutex<Option<String>>` (`pub(crate)` field, `parking_lot` is
already a dependency). A new `pub(crate) fn set_receipt_handle(&self, ..)`
replaces it.
- `SimpleConsumer::change_invisible_duration` writes the returned handle
back into the view when `ack_entry` is a `MessageView` (the `'static` bound
already allows the `Any` downcast). It still returns the new handle as before.
- The handle is only replaced on success. On error the view keeps its handle.
- Doc comment on `change_invisible_duration` describing this.
#### Notes for reviewers
- **No public API change.** `change_invisible_duration` keeps its signature
and return value; `MessageView` stays `Send + Sync`. A custom `AckMessageEntry`
implementation is not refreshed (the crate can not modify it); it can still use
the returned handle.
- **Scope.** Only the simple consumer. The push consumer never reuses a
message after nacking it, as in Java. #1381's
`LiteSimpleConsumer::change_invisible_duration` delegates to
`SimpleConsumer::change_invisible_duration` with a `&MessageView`, so it gets
the same behavior once both are merged.
- **Conflicts.** `git merge-tree` is clean against #1381, #1404, #1407,
#1409, #1411, #1414, #1416, #1420, #1421, #1423, #1425 and #1429. I also built
and tested the temporary merges where a PR touches `MessageView` or
`simple_consumer.rs` (the field type changes): `cargo test` passes with #1381
(99), #1404 (100), #1414 (94), #1420 (92), #1429 (99) and #1423 (110).
- It conflicts with #1427 in `simple_consumer.rs`: both change the body of
`change_invisible_duration` and both append tests. Resolution: keep #1427's
`to_pb_invisible_duration(..)?`, then this PR's write-back; keep all tests.
Verified on a temporary merge: `cargo test` 94 passed. Overview in #1412.
### How Did You Test This Change?
**Unit tests** in `simple_consumer.rs`:
- `change_invisible_duration_refreshes_receipt_handle`: change twice, then
ack the same view. The mocked client expects the handles `handle-1`,
`handle-2`, then the ack with `handle-3`. Fails on `master`: the view still
holds `handle-1` after the first change.
- `failed_change_invisible_duration_keeps_receipt_handle`: a failed change
leaves the handle unchanged.
```
$ cargo test -- --test-threads=1
test result: ok. 91 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-targets` reports the same findings as `master`.
`--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 and
group. Each message is received with a 60 s invisible duration; the same
program ran on `master` and on this branch:
| case | `master` | this PR |
|---|---|---|
| `change_invisible_duration(&view, 10s)` | `Ok`, `view.receipt_handle()`
unchanged | `Ok`, `view.receipt_handle()` equals the returned handle |
| then `ack(&view)`, observe 50 s | ack `Ok`, **redelivered** at ~40 s
(`delivery_attempt = 2`) | ack `Ok`, not redelivered |
| change twice, then `ack(&view)`, observe 50 s | all `Ok`, **redelivered**
at ~40 s | all `Ok`, not redelivered |
| change to 10 s, no ack | redelivered | redelivered (unchanged) |
Java 5.2.2 on the same server does not redeliver in the first case.
---
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]
