Copilot commented on code in PR #10787:
URL: https://github.com/apache/rocketmq/pull/10787#discussion_r3702242593
##########
proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java:
##########
@@ -306,7 +319,9 @@ public CompletableFuture<PopResult> popMessage(ProxyContext
ctx, AddressableMess
messageExt.getProperties().computeIfAbsent(MessageConst.PROPERTY_FIRST_POP_TIME,
k -> String.valueOf(responseHeader.getPopTime()));
messageExt.setBrokerName(messageQueue.getBrokerName());
messageExt.setTopic(messageQueue.getTopic());
+ validMessageExtList.add(messageExt);
}
+ popResult.setMsgFoundList(validMessageExtList);
Review Comment:
After filtering malformed entries, `popResult` can end up with
`PopStatus.FOUND` but an empty `msgFoundList` (the new test asserts this). This
is an inconsistent API signal for callers that treat FOUND as 'messages
returned' and can cause tight polling loops. Consider updating the status
(e.g., to `NO_NEW_MSG`) when `validMessageExtList` is empty, or
documenting/enforcing an invariant between status and message list.
##########
proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java:
##########
@@ -285,14 +286,26 @@ public CompletableFuture<PopResult>
popMessage(ProxyContext ctx, AddressableMess
} else {
if
(messageExt.getProperty(MessageConst.PROPERTY_POP_CK) == null) {
String key =
ExtraInfoUtil.getStartOffsetInfoMapKey(messageExt.getTopic(),
messageExt.getQueueId());
- int index =
sortMap.get(key).indexOf(messageExt.getQueueOffset());
- Long msgQueueOffset =
msgOffsetInfo.get(key).get(index);
+ List<Long> sortQueueOffsets = sortMap.get(key);
+ List<Long> msgQueueOffsets = msgOffsetInfo == null
? null : msgOffsetInfo.get(key);
+ Long startOffset = startOffsetInfo.get(key);
Review Comment:
The local variable name `startOffset` is ambiguous here (it can be confused
with other start offsets in POP flow or outer-scope variables). Renaming to
something keyed (e.g., `startOffsetForQueue` / `startOffsetForKey`) would make
the metadata usage clearer and reduce maintenance risk.
##########
proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalMessageServiceTest.java:
##########
@@ -340,6 +340,47 @@ public void testPopMessageWriteAndFlush() throws Exception
{
}
}
+ @Test
+ public void testPopMessageShouldSkipMessageWithMissingOffsetMetadata()
throws Exception {
Review Comment:
This test covers missing `msgOffsetInfo`, but the production changes also
handle invalid offset indexes (`index < 0` / `index >= msgQueueOffsets.size()`)
and missing `startOffsetInfo` entries for a key. Adding targeted tests for
those branches would prevent regressions in the new guard logic.
##########
proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java:
##########
@@ -285,14 +286,26 @@ public CompletableFuture<PopResult>
popMessage(ProxyContext ctx, AddressableMess
} else {
if
(messageExt.getProperty(MessageConst.PROPERTY_POP_CK) == null) {
String key =
ExtraInfoUtil.getStartOffsetInfoMapKey(messageExt.getTopic(),
messageExt.getQueueId());
- int index =
sortMap.get(key).indexOf(messageExt.getQueueOffset());
- Long msgQueueOffset =
msgOffsetInfo.get(key).get(index);
+ List<Long> sortQueueOffsets = sortMap.get(key);
+ List<Long> msgQueueOffsets = msgOffsetInfo == null
? null : msgOffsetInfo.get(key);
+ Long startOffset = startOffsetInfo.get(key);
+ if (sortQueueOffsets == null || msgQueueOffsets ==
null || startOffset == null) {
+ log.warn("Pop response offset metadata is
missing, key:{}", key);
+ continue;
+ }
+ int index =
sortQueueOffsets.indexOf(messageExt.getQueueOffset());
+ if (index < 0 || index >= msgQueueOffsets.size()) {
+ log.warn("Pop response offset metadata index
is invalid, key:{}, index:{}, msgOffsetCount:{}",
+ key, index, msgQueueOffsets.size());
+ continue;
+ }
Review Comment:
These `warn` logs execute per message and could become noisy under
persistent malformed metadata (especially with batch POP). Consider aggregating
(log once per response with counts/first key), rate-limiting, or downgrading to
`info/debug` while keeping enough context to diagnose.
--
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]
