leizhiyuan opened a new issue, #1434: URL: https://github.com/apache/rocketmq-clients/issues/1434
### 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` + #1381 (LiteSimpleConsumer) 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 `LiteSimpleConsumer::change_invisible_duration` (added by #1381) sends the lite topic but never sets `ChangeInvisibleDurationRequest.suspend`. The Java client sets `suspend = true` for a lite message (`ConsumerImpl#wrapChangeInvisibleDuration`, since #1206): ```java if (isLiteConsumer()) { if (messageView.getLiteTopic().isPresent()) { builder.setLiteTopic(messageView.getLiteTopic().get()); builder.setSuspend(true); } } ``` The broker handles the two cases differently (`ChangeInvisibleTimeProcessor#processChangeInvisibleTimeForLite`): - `suspend = true`: `suspendQueue(lmq, group, ...)`. The whole lite topic queue of the group stays invisible until the new time, and the delivery is not counted (`offsetConsumedCount` is decremented). - otherwise: `updateNextVisibleTime(...)` for this single offset only. **Scenario.** A lite simple consumer receives a message of one lite topic (e.g. one chat session or one device) and can not process it yet, so it calls `change_invisible_duration` to retry later. With Java the whole lite topic pauses, so the following messages of the same lite topic are not handed out in the meantime and order is kept. With Rust only this message is postponed, the broker treats it as a normal invisible-time change, and the call also counts towards the delivery attempts. **About `LitePushConsumer`.** Java's `LitePushConsumer` also goes through `wrapChangeInvisibleDuration` when the standard (non-FIFO) consume service nacks a failed message, so it sends `suspend = true` there as well. I tried the same on the Rust standard worker against a real server, and it shows why that path is risky: - The listener failed while `delivery_attempt <= 2`. - Without `suspend` (today): the message is redelivered with an increasing attempt and then succeeds. - With `suspend = true`: the attempt is no longer counted, and the message stayed at `delivery_attempt = 2`. It was redelivered every ~1.1 s for the whole 125 s observation window, about 114 times, together with the later messages of the same lite topic. It never reached `max_attempts` or the DLQ. So a message that keeps failing would loop forever instead of being dead-lettered. This server pushes `fifo = true`, so Java takes the FIFO local-retry path and does not reach this code; I could not compare it there. Whether a FAILURE nack should suspend is a question for the maintainers. The fix below therefore only covers the explicit `LiteSimpleConsumer::change_invisible_duration`. ### Steps to Reproduce Code review: `Client::change_invisible_duration_inner` builds the request with `suspend: None` for every client type (`rust/src/client.rs`). This can not be reproduced end to end on the server I use, because it rejects the `LITE_SIMPLE_CONSUMER` client type (`Client type is not consumer, type: LITE_SIMPLE_CONSUMER`), for Java 5.2.2 as well. ### What Did You Expect to See? `LiteSimpleConsumer::change_invisible_duration` of a lite message sends `suspend = true`, as the Java `LiteSimpleConsumer` does. ### What Did You See Instead? `suspend` is never set. ### Additional Context Found while comparing the Rust and Java SimpleConsumer implementations. **Maintainers, please confirm** the scope: (1) `suspend = true` for `LiteSimpleConsumer::change_invisible_duration`, as in Java; (2) the `LitePushConsumer` FAILURE nack keeps `suspend` unset, unlike Java, because of the endless redelivery shown above. --- Tracked in #1412. -- 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]
