leizhiyuan opened a new issue, #1410: URL: https://github.com/apache/rocketmq-clients/issues/1410
### Before Creating the Bug Report - [x] I found a bug, not just asking a question, which should be created in [GitHub Discussions](https://github.com/apache/rocketmq-clients/discussions). - [x] I have searched the [GitHub Issues](https://github.com/apache/rocketmq-clients/issues) and [GitHub Discussions](https://github.com/apache/rocketmq-clients/discussions) of this repository and believe that this is not a duplicate. - [x] I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ. ### Programming Language of the Client Rust ### Runtime Platform Environment macOS 27 (arm64) ### RocketMQ Version of the Client/Server Client: Rust `rocketmq` 5.0.0, master `a3d1de8b` Server: a managed RocketMQ 5.x service (Tencent Cloud TDMQ for RocketMQ 5), gRPC proxy endpoint ### Run or Compiler Version rustc 1.98.0 ### Describe the Bug If the `MessageListener` of a `PushConsumer` (or `LitePushConsumer`) panics, the panic unwinds through the consumer worker task and terminates it. The task is never restarted, and the message being consumed is neither acked nor nacked. **Scenario.** A listener hits a bug on one bad message: an `unwrap()` on malformed payload, an index out of bounds, and so on. The application expects this message to be treated as a consumption failure and retried, and expects the other messages to keep flowing. This is what happens with the Java client. With the Rust client: - The worker task that called the listener exits. Each message queue only has a fixed number of workers (`consumer_worker_count`), and nothing respawns them. So after a few such panics, a queue silently stops being consumed, while the process stays healthy. - The panicking message is not handed to `change_invisible_duration` (standard) or to the local retry / DLQ path (FIFO). It only comes back after its invisible duration expires. - With `LitePushConsumer`, all lite topics share the bind topic's queues, so a panic caused by one lite topic affects others. The listener is called directly, without any protection, in both workers: ```rust // rust/src/push_consumer.rs, StandardConsumerWorker::receive_messages let consume_result = (self.message_listener)(&message); // rust/src/push_consumer.rs, FifoConsumerWorker::consume_message_sequentially_static let consume_result = (message_listener)(&message); ``` **Java behavior that Rust should align with.** `ConsumeTask#call` wraps the listener: ```java try { consumeResult = messageListener.consume(messageView); } catch (Throwable t) { log.error("Message listener raised an exception while consuming messages, ...", t); // If exception was thrown during the period of message consumption, mark it as failure. consumeResult = ConsumeResult.FAILURE; throwable = t; } ``` The message then goes through the normal failure path: redelivery for standard consumption, local retry and then DLQ for FIFO. The consumer keeps running. ### Steps to Reproduce 1. Start a `LitePushConsumer` (or `PushConsumer`) whose listener panics on the first delivery of a specific message, e.g. `if body == "PANIC" && msg.delivery_attempt() <= 1 { panic!() }`. 2. Send that message, and a few seconds later send a normal message to the same lite topic. ### What Did You Expect to See? As with the Java client: the panicking message is treated as `FAILURE` and redelivered (attempt 2 succeeds), the following message is consumed, and an error is logged. ### What Did You See Instead? In our test, the panicking message was delivered twice with attempt `#1`, and its retry never succeeded. The following message on the same lite topic was not consumed within 60 s. Each panic removed one consumer worker for good. ### Additional Context Catching the panic does not work when the application is built with `panic = "abort"`. In that case the process aborts, which is the expected behavior of that setting. -- 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]
