leizhiyuan opened a new pull request, #1466: URL: https://github.com/apache/rocketmq-clients/pull/1466
### Which Issue(s) This PR Fixes Fixes #1464 ### Brief Description `fifoConsumeService` and `liteFifoConsumeService` went on with the next message as soon as `eraseFifoMessage` returned. For a failed message `eraseFifoMessage` only schedules the local redelivery, and the ack / suspend / DLQ forward are asynchronous too, so the following messages of the same message group or lite topic were consumed and acked before the failed one was redelivered. - `ProcessQueue#eraseFifoMessage` takes a `done` callback. It is called once the message is settled: after the local redeliveries (passed on recursively), then the ack, the suspend or the DLQ forward, right after the cache is evicted. - Both FIFO consume services consume the next message from `done`. This matches Java, where `consumeIteratively` continues in a listener of the future returned by `ProcessQueueImpl#eraseFifoMessage`. - Suspend in `liteFifoConsumeService`: the other messages of the same lite topic in the batch are still suspended right away; the messages of other lite topics continue once the suspended message is settled, like `LiteFifoConsumeService` in Java. #### Notes for reviewers - `ProcessQueue` is an unexported interface; `defaultProcessQueue` is its only implementation. - The settle callbacks are always invoked: ack and change invisible duration retry until they succeed or give up on `INVALID_RECEIPT_HANDLE` (callback with an error), and the DLQ forward retries until it succeeds. So `done` is called on every path, as Java's future completes. - Messages of different groups in a batch are still consumed in parallel when the FIFO consume accelerator is on. ### How Did You Test This Change? **Unit tests** (new `fifo_consume_service_test.go`; the consume service and the process queue run for real, only the RPCs are mocked): - `TestFifoConsumeServiceRetriesFailedMessageBeforeNextMessage`: fifo / lite fifo, with and without the accelerator. Message `a0` fails twice in a batch interleaved with another group. Expected order `a0@1 a0@2 a0@3 a1 a2`. On `master` all 4 subtests fail with `a0@1 a1@1 a2@1 a0@2 a0@3`. - `TestFifoConsumeServiceForwardsToDeadLetterQueueBeforeNextMessage`: `a0` runs out of attempts and goes to the DLQ before `a1` is consumed. Fails on `master`. - `TestLiteFifoConsumeServiceSuspendSkipsSameLiteTopic`: a suspend suspends the rest of the same lite topic and continues with the other lite topic. Passes before and after. ``` $ go test -race -count=5 -run FifoConsumeService . # ok $ go test -race ./... # ok $ go vet ./... # clean ``` **End-to-end** with a `LitePushConsumer` on a RocketMQ 5.x server with lite topic support (FIFO group, retry backoff 1s), 4 messages to one lite topic, `c2-0` fails twice: | | consume sequence | |---|---| | `master` | `c2-0(1,FAIL)@0s c2-1@0s [email protected] [email protected] c2-0(2,FAIL)@1s c2-0(3,OK)@2s` | | this PR | `c2-0(1,FAIL)@0s c2-0(2,FAIL)@1s c2-0(3,OK)@2s [email protected] [email protected] [email protected]` | | Java 5.2.2 | `c2-0(1,FAIL)@0s c2-0(2,FAIL)@1s c2-0(3,OK)@2s c2-1 c2-2 c2-3` | Listener that panics once on `c4-0`: `master` `c4-0, c4-1, c4-0` → this PR `c4-0, c4-0, c4-1`. Order per lite topic with 3 lite topics (accelerator on/off) and suspend (redelivered after 5.2s, other lite topics not affected) behave as before. -- 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]
