leizhiyuan opened a new issue, #1464: URL: https://github.com/apache/rocketmq-clients/issues/1464
### 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 Go ### Runtime Platform Environment macOS 27.0.1 arm64 ### RocketMQ Version of the Client/Server Client: Go master `e90ece7a` Server: RocketMQ 5.x proxy with lite topic support (FIFO consumer group, the server pushes `fifo=true`, retry max attempts 17, backoff 1s) ### Run or Compiler Version go1.27.0 darwin/arm64 ### Describe the Bug When a FIFO message fails (the listener returns `FAILURE` or panics), `PushConsumer` and `LitePushConsumer` consume the following messages of the same message group / lite topic **before** the failed message is redelivered. The FIFO order is broken, and the later messages are even acked before the failed one. `fifoConsumeService#consumeIteratively` and `liteFifoConsumeService#consumeIteratively` move on to the next message right after `eraseFifoMessage` returns ([consumer_service.go#L217-L218](https://github.com/apache/rocketmq-clients/blob/e90ece7a/golang/consumer_service.go#L217-L218), [#L275-L292](https://github.com/apache/rocketmq-clients/blob/e90ece7a/golang/consumer_service.go#L275-L292)): ```go pq.eraseFifoMessage(mv, result) fcs.consumeIteratively(pq, messageViewsPtr, ptr+1) ``` But for a failed message `eraseFifoMessage` only schedules the local redelivery with `consumeWithDuration` and returns at once ([process_queue.go#L91-L94](https://github.com/apache/rocketmq-clients/blob/e90ece7a/golang/process_queue.go#L91-L94)). The ack, the change of invisible duration and the forward to the DLQ are asynchronous too. Java waits: `FifoConsumeService#consumeIteratively` and `LiteFifoConsumeService#consumeIteratively` consume the next message in a listener of the future returned by `ProcessQueueImpl#eraseFifoMessage`, and that future completes only after the redelivery (recursively), the ack, the suspend or the DLQ forward is done. `LitePushConsumer` always uses the FIFO consume service, so every lite topic is affected. ### Steps to Reproduce `LitePushConsumer` bound to a lite topic, subscribe one lite topic, send 4 messages `c2-0..c2-3` to it, and let the listener return `FAILURE` twice for `c2-0`. The same happens with a normal FIFO `PushConsumer` and with a listener that panics. ### What Did You Expect to See? The same as the Java client (5.2.2, same server, same group): ``` c2-0(att=1,FAIL)@+0.00s c2-0(att=2,FAIL)@+1.00s c2-0(att=3,OK)@+2.00s c2-1(att=1,OK) c2-2(att=1,OK) c2-3(att=1,OK) ``` ### What Did You See Instead? ``` c2-0(att=1,FAIL)@+0.00s c2-1(att=1,OK)@+0.00s c2-2(att=1,OK)@+0.05s c2-3(att=1,OK)@+0.10s c2-0(att=2,FAIL)@+1.00s c2-0(att=3,OK)@+2.00s ``` `c2-1..c2-3` are consumed and acked while `c2-0` is still waiting for its redelivery. With a listener that panics once on `c4-0`, the order is `c4-0, c4-1, c4-0` (Java: `c4-0, c4-0, c4-1`). ### Additional Context The suspend path of `liteFifoConsumeService` (`ConsumerResultSuspend`) suspends the rest of the lite topic in the batch synchronously in the loop, so it is not affected; it only needs to keep working after a fix. -- 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]
