leizhiyuan opened a new pull request, #1414:
URL: https://github.com/apache/rocketmq-clients/pull/1414
### Which Issue(s) This PR Fixes
Fixes #1413
### Brief Description
**Scenario.** With FIFO consumption, a failed message is retried locally by
the client. Business code that branches on `MessageView::delivery_attempt()`
(give up after N attempts, alert on the last one, ...) never triggers. The
listener sees the server-provided attempt, typically `1`, on every local retry.
The message is retried until `max_attempts` is exhausted, and all later
messages of the same message group / lite topic stay blocked behind it. Both
`PushConsumer` and `LitePushConsumer` are affected, since they share
`FifoConsumerWorker`.
The root cause is in
`FifoConsumerWorker::consume_message_sequentially_static`. It counted attempts
in a local variable that was never written back to the `MessageView` handed to
the listener.
This PR aligns the FIFO local retry with the Java client,
`ProcessQueueImpl#eraseFifoMessage`:
```java
int attempt = messageView.getDeliveryAttempt();
if (FAILURE && attempt < maxAttempts) {
nextAttemptDelay = retryPolicy.getNextAttemptDelay(attempt);
attempt = messageView.incrementAndGetDeliveryAttempt();
consume(messageView, nextAttemptDelay); // then eraseFifoMessage
again
} else ... forwardToDeadLetterQueue(messageView) // reports
messageView.getDeliveryAttempt()
```
- **`MessageView::increment_and_get_delivery_attempt()`** (`pub(crate)`) is
added, matching `MessageViewImpl#incrementAndGetDeliveryAttempt`.
- **`consume_message_sequentially_static` reads and updates the attempt on
the `MessageView` itself.** On failure with `attempt < max_attempts`, it takes
the delay of the current attempt, increments the attempt, sleeps, and consumes
again. Otherwise it forwards to the DLQ with the current attempt.
- **The standard (non-FIFO) path is unchanged.** It redelivers through the
server, which owns the attempt.
#### Behavior changes
These were measured by running the new tests against both the old and the
new code. Example: `max_attempts = 4`, a message delivered with attempt `1`,
and a listener that always fails.
| | before | after (= Java) | externally visible |
|---|---|---|---|
| attempts seen by the listener | `1, 1, 1, 1` | `1, 2, 3, 4` | yes (the
fix) |
| number of listener calls before the DLQ | 4 | 4 (unchanged) | — |
| `delivery_attempt` in `ForwardMessageToDeadLetterQueueRequest` | `5` (max
+ 1) | **`4`** | **yes** |
| delay before the first local retry | `next[1]` (delay of attempt 2) |
**`next[0]`** (delay of attempt 1) | **yes** |
The last two rows change what the server receives in the DLQ request and the
retry timing. Both now match Java. The local retry still stops at the same
point: the message reaches the DLQ after the same number of listener calls.
#### Notes for reviewers
- **Missing `delivery_attempt` from the server.** If the server omits
`delivery_attempt`, `MessageView::from_pb_message` maps it to `0`. The first
local retry then asks the retry policy for the delay of attempt `0`, which is
`0` with both policies. The old code incremented first, so it never asked for
attempt `0`. Java would fail here with `checkArgument(attempt > 0)`. Servers
always set this field on delivered messages, so no extra fallback is added.
This keeps the behavior identical to Java for every value the server actually
sends.
- **Merge order and rebase.** All the PRs touching
`rust/src/push_consumer.rs` are tracked in #1412. Checked with `git merge-tree`
against the current PR heads:
- #1381, #1404, #1409: merge cleanly.
- #1407: merges cleanly as text, but the 3 tests added here then need the
extra `filter_expression` argument of `receive_messages`.
- #1411: conflicts in `push_consumer.rs`. Both PRs touch the listener call
in `consume_message_sequentially_static`, and both append tests at the end of
`mod tests`. I resolved it in a scratch worktree by keeping `consume_message()`
from #1411 together with the attempt handling from this PR, and keeping all
tests from both. On the result, `cargo test --lib` gives 94 passed.
I suggest merging this after #1407 and #1411. I will rebase it then.
### How Did You Test This Change?
**Unit tests.** Three were added. The two FIFO ones fail on `master`.
- `test_fifo_local_retry_increments_delivery_attempt_until_dlq`:
`max_attempts = 4`, and the listener always fails. It asserts that the listener
saw `[1, 2, 3, 4]`, and that the DLQ request has `delivery_attempt == 4` and
`max_delivery_attempts == 4`. On `master` the listener sees `[1, 1, 1, 1]` and
the DLQ request carries `5`.
- `test_fifo_local_retry_succeeds_with_incremented_delivery_attempt`: the
listener fails twice and then succeeds. It asserts `[1, 2, 3]`, one ack and no
DLQ. It also asserts the run takes < 2 s, with `next = [10ms, 10ms, 5s]`, which
proves that the retries use the delays of attempts 1 and 2. On `master` it sees
`[1, 1, 1]` and waits the 5 s step.
- `test_standard_failure_keeps_server_delivery_attempt`: a regression guard
for the non-FIFO path. A message delivered with attempt `2` is seen as `2`. It
is nacked once via `change_invisible_duration` with the delay for attempt 2
(`7s`), and is neither acked nor sent to the DLQ. It passes before and after.
```
$ cargo test
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)
# full `cargo test` repeated 10 times and `cargo test --lib` 30 times: 0
failures
$ cargo fmt --check # clean
$ RUSTDOCFLAGS="--cfg docsrs" cargo doc --no-deps --all-features # no
warnings
$ cargo build --examples # ok
```
`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).
This PR adds no clippy findings in library code.
**End-to-end.** I ran this against a managed RocketMQ 5 server, with a
`LitePushConsumer` using `fifo = true`. The listener fails while
`delivery_attempt <= 1`. Three more messages were sent to the same lite topic
afterwards.
| | before | after |
|---|---|---|
| deliveries seen by the listener | `FAIL#1` × 5, the following messages
blocked | `FAIL#1`, `FAIL#2`, `next-0#1`, `next-1#1`, `next-2#1` |
| 30 messages on one lite topic (`fifo = true`) | in order | in order |
The e2e run was done on top of #1407. Without it, `LitePushConsumer` cannot
receive messages from this server at all.
---
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]
