leizhiyuan opened a new pull request, #1411:
URL: https://github.com/apache/rocketmq-clients/pull/1411
### Which Issue(s) This PR Fixes
Fixes #1410
### Brief Description
**Scenario.** A `MessageListener` panics on one bad message, e.g. an
`unwrap()` on a malformed payload. Today the panic unwinds through the consumer
worker task and ends it. Workers are never respawned, so every such panic
permanently removes one of the queue's `consumer_worker_count` workers, and
after a few panics the queue silently stops being consumed. The message being
consumed is neither acked nor nacked, so it only comes back once its invisible
duration expires. With `LitePushConsumer` all lite topics share the bind
topic's queues, so a panic caused by one lite topic affects the others.
The Java client handles this in `ConsumeTask#call`. It catches any
`Throwable` raised by the listener, logs it, and marks the consumption as
`ConsumeResult.FAILURE`. The message then follows the normal failure path, and
the consumer keeps running.
This PR does the same:
- **Adds `consume_message()`**, which calls the listener inside
`std::panic::catch_unwind`. On panic it logs an error with the topic, the
message id and the panic payload, then returns `ConsumeResult::FAILURE`.
- **Uses it at both listener call sites:**
- `StandardConsumerWorker::receive_messages`: the message is then
redelivered via `change_invisible_duration`.
- `FifoConsumerWorker::consume_message_sequentially_static`: local retry,
then DLQ. This path is also used by the FIFO consume accelerator.
- **Unwind safety.** `AssertUnwindSafe` is used because the listener is a
user-provided `Fn(&MessageView)`. After a panic, the client only reads the
`MessageView` it owns. Any state the listener itself captured is the
application's responsibility, as with a Java listener that throws.
- **`panic = "abort"` builds.** A panic cannot be caught there, so the
process aborts as configured. This is documented on `consume_message()`.
- **Default panic hook.** It is left untouched, so the usual `thread '...'
panicked at ...` line is still printed to stderr. A client library should not
override the application's global hook.
**Test fix in `client.rs`.** `client::tests::client_get_session` now takes
`MTX` like the other tests that set the global `MockSession::new` expectation.
Without the lock it could race with `test_get_session_for_lite_consumer`, which
then fails with `MockSession::new(?, ?, ?): No matching expectation found`. The
race was latent on `master`, with 0 failures in 25 runs. Adding tests changed
the scheduling enough to expose it, with roughly 1 failure in 4 to 8 runs. With
the lock: 0 failures in 30 runs.
### How Did You Test This Change?
**Unit tests.** Both fail without the fix, because the panic propagates out
of the worker:
- `test_standard_consume_messages_listener_panic`: the listener always
panics. The test asserts that `receive_messages` returns `Ok`,
`change_invisible_duration` is called once and `ack_message` is never called.
- `test_fifo_consume_message_listener_panic_then_success`: the listener
panics on the first call and returns `SUCCESS` on the retry. The test asserts
the listener ran twice, the message was acked once and nothing went to the DLQ.
```
$ cargo test
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="--cfg docsrs" cargo doc --no-deps --all-features # no
warnings
$ cargo build --examples # ok
```
`cargo clippy --all-features -- -D warnings` fails on `master` with 2
`for_kv_map` errors in `rust/src/client.rs` (lines 904 and 912). They are
unchanged here, and this PR adds no clippy findings in library code.
**End-to-end.** I ran this against a managed RocketMQ 5 server with a
`LitePushConsumer`. The listener panics on the first delivery of a message. A
normal message is sent to the same lite topic 5 s later, and another one to a
different lite topic.
| | before | after |
|---|---|---|
| deliveries seen by the listener (`body#attempt`) on the panicking lite
topic | `PANIC#1`, `PANIC#1`; the retry never succeeds | `PANIC#1` → `PANIC#2`
(succeeds) → `after-panic#1` |
| following message on the same lite topic within 60 s | not consumed |
consumed |
| client log | — | one `ERROR message listener panicked while consuming
message, treat it as failure, ...` |
The e2e run was done on top of #1407. Without it, `LitePushConsumer` cannot
receive messages from this server at all. This PR does not depend on #1407 or
#1409, but all three touch `rust/src/push_consumer.rs`, so whichever lands
later needs a trivial rebase. For example, these tests need the extra
`filter_expression` argument once #1407 is merged.
--
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]
