From ae63fe7da84d0ad249f646dce2759f923965a0a8 Mon Sep 17 00:00:00 2001 From: liuhy Date: Fri, 31 Jul 2026 05:45:16 -0700 Subject: [PATCH 1/3] [ISSUE #10730] Avoid raw local POP message logs --- .../service/message/LocalMessageService.java | 18 +++++++++++++- .../message/LocalMessageServiceTest.java | 24 +++++++++++++++++++ 2 files changed, 41 insertions(+), 1 deletion(-) diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java index c93fa93983c..469d411fe15 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java @@ -25,6 +25,7 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.TreeSet; import java.util.concurrent.CompletableFuture; import org.apache.commons.lang3.NotImplementedException; import org.apache.rocketmq.broker.BrokerController; @@ -288,7 +289,8 @@ public CompletableFuture popMessage(ProxyContext ctx, AddressableMess int index = sortMap.get(key).indexOf(messageExt.getQueueOffset()); Long msgQueueOffset = msgOffsetInfo.get(key).get(index); if (msgQueueOffset != messageExt.getQueueOffset()) { - log.warn("Queue offset [{}] of msg is strange, not equal to the stored in msg, {}", msgQueueOffset, messageExt); + log.warn("Queue offset [{}] of msg is strange, not equal to the stored in msg, msgSummary:{}", + msgQueueOffset, summarizeMessageExt(messageExt)); } messageExt.getProperties().put(MessageConst.PROPERTY_POP_CK, @@ -312,6 +314,20 @@ public CompletableFuture popMessage(ProxyContext ctx, AddressableMess }); } + static String summarizeMessageExt(MessageExt messageExt) { + if (messageExt == null) { + return "msg=null"; + } + return "topic=" + messageExt.getTopic() + + ", msgId=" + messageExt.getMsgId() + + ", queueId=" + messageExt.getQueueId() + + ", queueOffset=" + messageExt.getQueueOffset() + + ", commitLogOffset=" + messageExt.getCommitLogOffset() + + ", bodySize=" + (messageExt.getBody() == null ? 0 : messageExt.getBody().length) + + ", propertyKeys=" + (messageExt.getProperties() == null + ? "[]" : new TreeSet<>(messageExt.getProperties().keySet())); + } + @Override public CompletableFuture changeInvisibleTime(ProxyContext ctx, ReceiptHandle handle, String messageId, ChangeInvisibleTimeRequestHeader requestHeader, long timeoutMillis) { diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalMessageServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalMessageServiceTest.java index 52ba521f802..732c9b0779a 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalMessageServiceTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalMessageServiceTest.java @@ -136,6 +136,30 @@ public void setUp() throws Throwable { .withVal(ContextVariable.LOCAL_ADDRESS, "0.0.0.2"); } + @Test + public void testSummarizeMessageExtDoesNotExposeBodyOrPropertyValues() { + MessageExt messageExt = new MessageExt(); + messageExt.setTopic(topic); + messageExt.setMsgId("msgId"); + messageExt.setQueueId(1); + messageExt.setQueueOffset(2L); + messageExt.setCommitLogOffset(3L); + messageExt.setBody(new byte[] {115, 101, 99, 114, 101, 116}); + messageExt.putUserProperty("secretKey", "secretValue"); + + String summary = LocalMessageService.summarizeMessageExt(messageExt); + + assertThat(summary).contains("topic=" + topic); + assertThat(summary).contains("msgId=msgId"); + assertThat(summary).contains("queueId=1"); + assertThat(summary).contains("queueOffset=2"); + assertThat(summary).contains("commitLogOffset=3"); + assertThat(summary).contains("bodySize=6"); + assertThat(summary).contains("secretKey"); + assertThat(summary).doesNotContain("secretValue"); + assertThat(summary).doesNotContain("115, 101, 99, 114, 101, 116"); + } + @Test public void testSendMessageWriteAndFlush() throws Exception { Message message = new Message(topic, "body".getBytes(StandardCharsets.UTF_8)); From 4f967ae318ff49ce07021aaf550115c590b93b63 Mon Sep 17 00:00:00 2001 From: liuhy Date: Mon, 3 Aug 2026 00:46:46 -0700 Subject: [PATCH 2/3] [ISSUE #10786] Tolerate malformed local POP offset metadata --- .../service/message/LocalMessageService.java | 21 ++++++++-- .../message/LocalMessageServiceTest.java | 41 +++++++++++++++++++ 2 files changed, 59 insertions(+), 3 deletions(-) diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java index 469d411fe15..60d17c0b2dd 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java @@ -273,6 +273,7 @@ public CompletableFuture popMessage(ProxyContext ctx, AddressableMess sortMap.get(key).add(messageExt.getQueueOffset()); } Map map = new HashMap<>(5); + List validMessageExtList = new ArrayList<>(messageExtList.size()); for (MessageExt messageExt : messageExtList) { if (startOffsetInfo == null) { // we should set the check point info to extraInfo field , if the command is popMsg @@ -286,15 +287,27 @@ public CompletableFuture 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 sortQueueOffsets = sortMap.get(key); + List 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; + } + Long msgQueueOffset = msgQueueOffsets.get(index); if (msgQueueOffset != messageExt.getQueueOffset()) { log.warn("Queue offset [{}] of msg is strange, not equal to the stored in msg, msgSummary:{}", msgQueueOffset, summarizeMessageExt(messageExt)); } messageExt.getProperties().put(MessageConst.PROPERTY_POP_CK, - ExtraInfoUtil.buildExtraInfo(startOffsetInfo.get(key), responseHeader.getPopTime(), responseHeader.getInvisibleTime(), + ExtraInfoUtil.buildExtraInfo(startOffset, responseHeader.getPopTime(), responseHeader.getInvisibleTime(), responseHeader.getReviveQid(), messageExt.getTopic(), messageQueue.getBrokerName(), messageExt.getQueueId(), msgQueueOffset) ); if (requestHeader.isOrder() && orderCountInfo != null) { @@ -308,7 +321,9 @@ public CompletableFuture 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); } return popResult; }); diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalMessageServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalMessageServiceTest.java index 732c9b0779a..3169a2a6ca1 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalMessageServiceTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalMessageServiceTest.java @@ -364,6 +364,47 @@ public void testPopMessageWriteAndFlush() throws Exception { } } + @Test + public void testPopMessageShouldSkipMessageWithMissingOffsetMetadata() throws Exception { + int reviveQueueId = 1; + long popTime = System.currentTimeMillis(); + long invisibleTime = 3000L; + long startOffset = 100L; + StringBuilder startOffsetStringBuilder = new StringBuilder(); + ExtraInfoUtil.buildStartOffsetInfo(startOffsetStringBuilder, topic, queueId, startOffset); + MessageExt message = buildMessageExt(topic, queueId, startOffset); + byte[] body = MessageDecoder.encode(message, false); + PopMessageRequestHeader requestHeader = new PopMessageRequestHeader(); + requestHeader.setInvisibleTime(invisibleTime); + Mockito.when(popMessageProcessorMock.processRequest(Mockito.any(SimpleChannelHandlerContext.class), Mockito.argThat(argument -> { + boolean first = argument.getCode() == RequestCode.POP_MESSAGE; + boolean second = argument.readCustomHeader() instanceof PopMessageRequestHeader; + return first && second; + }))).thenAnswer(invocation -> { + SimpleChannelHandlerContext simpleChannelHandlerContext = invocation.getArgument(0); + RemotingCommand request = invocation.getArgument(1); + RemotingCommand response = RemotingCommand.createResponseCommand(PopMessageResponseHeader.class); + response.setOpaque(request.getOpaque()); + response.setCode(ResponseCode.SUCCESS); + response.setBody(body); + PopMessageResponseHeader responseHeader = (PopMessageResponseHeader) response.readCustomHeader(); + responseHeader.setStartOffsetInfo(startOffsetStringBuilder.toString()); + responseHeader.setInvisibleTime(requestHeader.getInvisibleTime()); + responseHeader.setPopTime(popTime); + responseHeader.setReviveQid(reviveQueueId); + simpleChannelHandlerContext.writeAndFlush(response); + return null; + }); + + MessageQueue messageQueue = new MessageQueue(topic, brokerName, queueId); + CompletableFuture future = localMessageService.popMessage(proxyContext, + new AddressableMessageQueue(messageQueue, ""), requestHeader, 1000L); + PopResult popResult = future.get(); + + assertThat(popResult.getPopStatus()).isEqualTo(PopStatus.FOUND); + assertThat(popResult.getMsgFoundList()).isEmpty(); + } + @Test public void testPopMessagePollingTimeout() throws Exception { RemotingCommand remotingCommand = RemotingCommand.createResponseCommand(ResponseCode.POLLING_TIMEOUT, ""); From 37115962cb877cd4b6a96e84a907f004848a00ac Mon Sep 17 00:00:00 2001 From: liuhy Date: Tue, 4 Aug 2026 03:40:19 -0700 Subject: [PATCH 3/3] fix(proxy): preserve local pop response invariants --- .../service/message/LocalMessageService.java | 32 ++++-- .../message/LocalMessageServiceTest.java | 99 +++++++++++++++---- 2 files changed, 105 insertions(+), 26 deletions(-) diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java index 60d17c0b2dd..a7cae600cd7 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java @@ -274,6 +274,10 @@ public CompletableFuture popMessage(ProxyContext ctx, AddressableMess } Map map = new HashMap<>(5); List validMessageExtList = new ArrayList<>(messageExtList.size()); + int missingOffsetMetadataCount = 0; + String firstMissingOffsetMetadataKey = null; + int invalidOffsetIndexCount = 0; + String firstInvalidOffsetIndexKey = null; for (MessageExt messageExt : messageExtList) { if (startOffsetInfo == null) { // we should set the check point info to extraInfo field , if the command is popMsg @@ -289,15 +293,20 @@ public CompletableFuture popMessage(ProxyContext ctx, AddressableMess String key = ExtraInfoUtil.getStartOffsetInfoMapKey(messageExt.getTopic(), messageExt.getQueueId()); List sortQueueOffsets = sortMap.get(key); List 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); + Long startOffsetForQueue = startOffsetInfo.get(key); + if (sortQueueOffsets == null || msgQueueOffsets == null || startOffsetForQueue == null) { + missingOffsetMetadataCount++; + if (firstMissingOffsetMetadataKey == null) { + firstMissingOffsetMetadataKey = 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()); + invalidOffsetIndexCount++; + if (firstInvalidOffsetIndexKey == null) { + firstInvalidOffsetIndexKey = key; + } continue; } Long msgQueueOffset = msgQueueOffsets.get(index); @@ -307,7 +316,7 @@ public CompletableFuture popMessage(ProxyContext ctx, AddressableMess } messageExt.getProperties().put(MessageConst.PROPERTY_POP_CK, - ExtraInfoUtil.buildExtraInfo(startOffset, responseHeader.getPopTime(), responseHeader.getInvisibleTime(), + ExtraInfoUtil.buildExtraInfo(startOffsetForQueue, responseHeader.getPopTime(), responseHeader.getInvisibleTime(), responseHeader.getReviveQid(), messageExt.getTopic(), messageQueue.getBrokerName(), messageExt.getQueueId(), msgQueueOffset) ); if (requestHeader.isOrder() && orderCountInfo != null) { @@ -323,7 +332,18 @@ public CompletableFuture popMessage(ProxyContext ctx, AddressableMess messageExt.setTopic(messageQueue.getTopic()); validMessageExtList.add(messageExt); } + if (missingOffsetMetadataCount > 0) { + log.warn("Skipped {} POP messages because offset metadata is missing, first key:{}", + missingOffsetMetadataCount, firstMissingOffsetMetadataKey); + } + if (invalidOffsetIndexCount > 0) { + log.warn("Skipped {} POP messages because offset metadata index is invalid, first key:{}", + invalidOffsetIndexCount, firstInvalidOffsetIndexKey); + } popResult.setMsgFoundList(validMessageExtList); + if (validMessageExtList.isEmpty() && !messageExtList.isEmpty()) { + popResult.setPopStatus(PopStatus.NO_NEW_MSG); + } } return popResult; }); diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalMessageServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalMessageServiceTest.java index 3169a2a6ca1..8e36ebd483f 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalMessageServiceTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalMessageServiceTest.java @@ -376,35 +376,70 @@ public void testPopMessageShouldSkipMessageWithMissingOffsetMetadata() throws Ex byte[] body = MessageDecoder.encode(message, false); PopMessageRequestHeader requestHeader = new PopMessageRequestHeader(); requestHeader.setInvisibleTime(invisibleTime); - Mockito.when(popMessageProcessorMock.processRequest(Mockito.any(SimpleChannelHandlerContext.class), Mockito.argThat(argument -> { - boolean first = argument.getCode() == RequestCode.POP_MESSAGE; - boolean second = argument.readCustomHeader() instanceof PopMessageRequestHeader; - return first && second; - }))).thenAnswer(invocation -> { - SimpleChannelHandlerContext simpleChannelHandlerContext = invocation.getArgument(0); - RemotingCommand request = invocation.getArgument(1); - RemotingCommand response = RemotingCommand.createResponseCommand(PopMessageResponseHeader.class); - response.setOpaque(request.getOpaque()); - response.setCode(ResponseCode.SUCCESS); - response.setBody(body); - PopMessageResponseHeader responseHeader = (PopMessageResponseHeader) response.readCustomHeader(); - responseHeader.setStartOffsetInfo(startOffsetStringBuilder.toString()); - responseHeader.setInvisibleTime(requestHeader.getInvisibleTime()); - responseHeader.setPopTime(popTime); - responseHeader.setReviveQid(reviveQueueId); - simpleChannelHandlerContext.writeAndFlush(response); - return null; - }); + mockPopMessageResponse(body, startOffsetStringBuilder.toString(), null, popTime, + requestHeader.getInvisibleTime(), reviveQueueId); MessageQueue messageQueue = new MessageQueue(topic, brokerName, queueId); CompletableFuture future = localMessageService.popMessage(proxyContext, new AddressableMessageQueue(messageQueue, ""), requestHeader, 1000L); PopResult popResult = future.get(); - assertThat(popResult.getPopStatus()).isEqualTo(PopStatus.FOUND); + assertThat(popResult.getPopStatus()).isEqualTo(PopStatus.NO_NEW_MSG); + assertThat(popResult.getMsgFoundList()).isEmpty(); + } + + @Test + public void testPopMessageShouldSkipMessageWithMissingStartOffset() throws Exception { + int reviveQueueId = 1; + long popTime = System.currentTimeMillis(); + long invisibleTime = 3000L; + long startOffset = 100L; + MessageExt message = buildMessageExt(topic, queueId, startOffset); + StringBuilder startOffsetInfo = new StringBuilder(); + StringBuilder msgOffsetInfo = new StringBuilder(); + ExtraInfoUtil.buildStartOffsetInfo(startOffsetInfo, topic, queueId + 1, startOffset); + ExtraInfoUtil.buildMsgOffsetInfo(msgOffsetInfo, topic, queueId, Collections.singletonList(startOffset)); + PopMessageRequestHeader requestHeader = new PopMessageRequestHeader(); + requestHeader.setInvisibleTime(invisibleTime); + mockPopMessageResponse(MessageDecoder.encode(message, false), startOffsetInfo.toString(), msgOffsetInfo.toString(), + popTime, invisibleTime, reviveQueueId); + + PopResult popResult = localMessageService.popMessage(proxyContext, + new AddressableMessageQueue(new MessageQueue(topic, brokerName, queueId), ""), requestHeader, 1000L).get(); + + assertThat(popResult.getPopStatus()).isEqualTo(PopStatus.NO_NEW_MSG); assertThat(popResult.getMsgFoundList()).isEmpty(); } + @Test + public void testPopMessageShouldSkipMessageWithInvalidOffsetIndex() throws Exception { + int reviveQueueId = 1; + long popTime = System.currentTimeMillis(); + long invisibleTime = 3000L; + long startOffset = 100L; + MessageExt firstMessage = buildMessageExt(topic, queueId, startOffset); + MessageExt secondMessage = buildMessageExt(topic, queueId, startOffset + 1); + byte[] firstMessageBody = MessageDecoder.encode(firstMessage, false); + byte[] secondMessageBody = MessageDecoder.encode(secondMessage, false); + ByteBuffer body = ByteBuffer.allocate(firstMessageBody.length + secondMessageBody.length); + body.put(firstMessageBody).put(secondMessageBody); + StringBuilder startOffsetInfo = new StringBuilder(); + StringBuilder msgOffsetInfo = new StringBuilder(); + ExtraInfoUtil.buildStartOffsetInfo(startOffsetInfo, topic, queueId, startOffset); + ExtraInfoUtil.buildMsgOffsetInfo(msgOffsetInfo, topic, queueId, Collections.singletonList(startOffset)); + PopMessageRequestHeader requestHeader = new PopMessageRequestHeader(); + requestHeader.setInvisibleTime(invisibleTime); + mockPopMessageResponse(body.array(), startOffsetInfo.toString(), msgOffsetInfo.toString(), popTime, + invisibleTime, reviveQueueId); + + PopResult popResult = localMessageService.popMessage(proxyContext, + new AddressableMessageQueue(new MessageQueue(topic, brokerName, queueId), ""), requestHeader, 1000L).get(); + + assertThat(popResult.getPopStatus()).isEqualTo(PopStatus.FOUND); + assertThat(popResult.getMsgFoundList()).hasSize(1); + assertThat(popResult.getMsgFoundList().get(0).getQueueOffset()).isEqualTo(startOffset); + } + @Test public void testPopMessagePollingTimeout() throws Exception { RemotingCommand remotingCommand = RemotingCommand.createResponseCommand(ResponseCode.POLLING_TIMEOUT, ""); @@ -542,6 +577,30 @@ private MessageExt buildMessageExt(String topic, int queueId, long queueOffset) return message1; } + private void mockPopMessageResponse(byte[] body, String startOffsetInfo, String msgOffsetInfo, long popTime, + long invisibleTime, int reviveQueueId) throws RemotingCommandException { + Mockito.when(popMessageProcessorMock.processRequest(Mockito.any(SimpleChannelHandlerContext.class), Mockito.argThat(argument -> { + boolean first = argument.getCode() == RequestCode.POP_MESSAGE; + boolean second = argument.readCustomHeader() instanceof PopMessageRequestHeader; + return first && second; + }))).thenAnswer(invocation -> { + SimpleChannelHandlerContext simpleChannelHandlerContext = invocation.getArgument(0); + RemotingCommand request = invocation.getArgument(1); + RemotingCommand response = RemotingCommand.createResponseCommand(PopMessageResponseHeader.class); + response.setOpaque(request.getOpaque()); + response.setCode(ResponseCode.SUCCESS); + response.setBody(body); + PopMessageResponseHeader responseHeader = (PopMessageResponseHeader) response.readCustomHeader(); + responseHeader.setStartOffsetInfo(startOffsetInfo); + responseHeader.setMsgOffsetInfo(msgOffsetInfo); + responseHeader.setInvisibleTime(invisibleTime); + responseHeader.setPopTime(popTime); + responseHeader.setReviveQid(reviveQueueId); + simpleChannelHandlerContext.writeAndFlush(response); + return null; + }); + } + private void assertMessageExt(MessageExt messageExt1, MessageExt messageExt2) { assertThat(messageExt1.getBody()).isEqualTo(messageExt2.getBody()); assertThat(messageExt1.getTopic()).isEqualTo(messageExt2.getTopic());