R0CKing666 opened a new issue, #11195:
URL: https://github.com/apache/rocketmq/issues/11195
### Which version of Apache RocketMQ
5.5.1 (develop)
### Description
`PopConsumerService.popAsync` (the RocksDB/KV POP path, enabled by
`popConsumerKVServiceEnable=true`) fails to recode POP retry messages back to
their origin topic because the recode branch has three defects:
1. **Inverted gate** — `recode = isPopResponseReturnActualRetryTopic()` is
missing the `!`. With the default `popResponseReturnActualRetryTopic=false` the
branch never executes; with `true` it executes when it should not. This
contradicts the code's own comment and the non-KV `PopMessageProcessor` path,
which recodes only when `!isPopResponseReturnActualRetryTopic() && isRetry`.
2. **Wrong origin topic** — `recodeRetryMessage(getMessageResult,
popConsumerRecord.getTopicId(), ...)` passes the *retry* topic (e.g.
`%RETRY%group_topic`) as the target, so `messageExt.setTopic(topicId)` inside
`recodeRetryMessage` is a no-op. The business topic should be passed.
3. **Wrong checkpoint offset** — `recodeRetryMessage(...,
popConsumerRecord.getQueueId(), ...)` passes the queue index where the message
queue offset is expected, corrupting the `POP_CK` checkpoint (`ckQueueOffset`)
used for ack / offset advancement.
### Expected behavior
With the default `popResponseReturnActualRetryTopic=false`, a retry message
returned by POP should be recoded back to its origin topic, matching the non-KV
path.
### How to reproduce
Enable `popConsumerKVServiceInit=true` and
`popConsumerKVServiceEnable=true`, write a message into the retry topic, POP it
and inspect the response buffer. With the default config the returned message
keeps the `%RETRY%...` topic (no recode) instead of the origin topic; with
`popResponseReturnActualRetryTopic=true` it is recoded (opposite of the
expected behavior).
### Proposed fix
```java
boolean recode = !brokerConfig.isPopResponseReturnActualRetryTopic();
if (recode && popConsumerRecord.isRetry()) {
result.getGetMessageResultList().set(i, this.recodeRetryMessage(
getMessageResult, topicId,
popConsumerRecord.getOffset(), result.getPopTime(), invisibleTime));
}
```
--
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]
