Skip to content

[ISSUE #10943] Tolerate incomplete queue-offset metadata during message decoding #10944

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Aias00 wants to merge 2 commits into
base: develop
Choose a base branch
from
Open

[ISSUE #10943] Tolerate incomplete queue-offset metadata during message decoding #10944

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
69b3a66
fix(broker): tolerate missing escape bridge queue offsets
Aias00 Aug 15, 2026
e79b65e
fix(broker): tolerate missing revive queue offsets
Aias00 Aug 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions broker/src/main/java/org/apache/rocketmq/broker/failover/EscapeBridge.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -342,6 +342,7 @@ protected List<MessageExt> decodeMsgList(GetMessageResult getMessageResult, bool
List<MessageExt> foundList = new ArrayList<>();
try {
List<ByteBuffer> messageBufferList = getMessageResult.getMessageBufferList();
List<Long> messageQueueOffsets = getMessageResult.getMessageQueueOffset();
if (messageBufferList != null) {
for (int i = 0; i < messageBufferList.size(); i++) {
ByteBuffer bb = messageBufferList.get(i);
Expand All @@ -354,8 +355,10 @@ protected List<MessageExt> decodeMsgList(GetMessageResult getMessageResult, bool
LOG.error("decode msgExt is null {}", getMessageResult);
continue;
}
// use CQ offset, not offset in Message
msgExt.setQueueOffset(getMessageResult.getMessageQueueOffset().get(i));
// Use the CQ offset when the store supplied it; otherwise retain the decoded offset.
if (messageQueueOffsets != null && i < messageQueueOffsets.size()) {
msgExt.setQueueOffset(messageQueueOffsets.get(i));
}
foundList.add(msgExt);
}
}
Expand Down
5 changes: 4 additions & 1 deletion broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -309,6 +309,7 @@ private List<MessageExt> decodeMsgList(GetMessageResult getMessageResult, boolea
List<MessageExt> foundList = new ArrayList<>();
try {
List<ByteBuffer> messageBufferList = getMessageResult.getMessageBufferList();
List<Long> messageQueueOffsets = getMessageResult.getMessageQueueOffset();
if (messageBufferList != null) {
for (int i = 0; i < messageBufferList.size(); i++) {
ByteBuffer bb = messageBufferList.get(i);
Expand All @@ -322,7 +323,9 @@ private List<MessageExt> decodeMsgList(GetMessageResult getMessageResult, boolea
continue;
}
// use CQ offset, not offset in Message
msgExt.setQueueOffset(getMessageResult.getMessageQueueOffset().get(i));
if (messageQueueOffsets != null && i < messageQueueOffsets.size()) {
msgExt.setQueueOffset(messageQueueOffsets.get(i));
}
foundList.add(msgExt);
}
}
Expand Down
17 changes: 17 additions & 0 deletions broker/src/test/java/org/apache/rocketmq/broker/failover/EscapeBridgeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -353,6 +353,23 @@ public void decodeMsgListTest_messageNotNull() throws Exception {
Assert.assertTrue(Arrays.equals(msg.getBody(), list.get(0).getBody()));
}

@Test
public void decodeMsgListTest_messageOffsetMissing() throws Exception {
MessageExt msg = new MessageExt();
msg.setBody("HW".getBytes());
msg.setTopic("topic");
msg.setBornHost(new InetSocketAddress("127.0.0.1", 9000));
msg.setStoreHost(new InetSocketAddress("127.0.0.1", 9000));
ByteBuffer byteBuffer = ByteBuffer.wrap(MessageDecoder.encode(msg, false));
SelectMappedBufferResult result = new SelectMappedBufferResult(0, byteBuffer, 10, new DefaultMappedFile());

getMessageResult.addMessage(result);

List<MessageExt> list = escapeBridge.decodeMsgList(getMessageResult, false);
Assert.assertEquals(1, list.size());
Assert.assertTrue(Arrays.equals(msg.getBody(), list.get(0).getBody()));
}

@Test
public void testPutMessageToRemoteBroker_noSpecificBrokerName_hasRemoteBroker() throws Exception {
MessageExtBrokerInner message = new MessageExtBrokerInner();
Expand Down
27 changes: 27 additions & 0 deletions broker/src/test/java/org/apache/rocketmq/broker/processor/PopReviveServiceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,9 +38,12 @@
import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.store.AppendMessageResult;
import org.apache.rocketmq.store.AppendMessageStatus;
import org.apache.rocketmq.store.GetMessageResult;
import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.PutMessageResult;
import org.apache.rocketmq.store.PutMessageStatus;
import org.apache.rocketmq.store.SelectMappedBufferResult;
import org.apache.rocketmq.store.logfile.DefaultMappedFile;
import org.apache.rocketmq.store.pop.AckMsg;
import org.apache.rocketmq.store.pop.BatchAckMsg;
import org.apache.rocketmq.store.pop.PopCheckPoint;
Expand All @@ -53,7 +56,10 @@
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;

import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
Expand Down Expand Up @@ -137,6 +143,27 @@ public void before() {
popReviveService.setShouldRunPopRevive(true);
}

@Test
@SuppressWarnings("unchecked")
public void testDecodeMsgListWithoutQueueOffsets() throws Exception {
MessageExt message = new MessageExt();
message.setBody("message".getBytes());
message.setTopic(TOPIC);
message.setBornHost(new InetSocketAddress("127.0.0.1", 9000));
message.setStoreHost(new InetSocketAddress("127.0.0.1", 9000));

GetMessageResult getMessageResult = new GetMessageResult();
ByteBuffer buffer = ByteBuffer.wrap(MessageDecoder.encode(message, false));
getMessageResult.addMessage(new SelectMappedBufferResult(0, buffer, buffer.remaining(), new DefaultMappedFile()));

Method decodeMsgList = PopReviveService.class.getDeclaredMethod("decodeMsgList", GetMessageResult.class, boolean.class);
decodeMsgList.setAccessible(true);
List<MessageExt> decodedMessages = (List<MessageExt>) decodeMsgList.invoke(popReviveService, getMessageResult, false);

Assert.assertEquals(1, decodedMessages.size());
Assert.assertArrayEquals(message.getBody(), decodedMessages.get(0).getBody());
}

@Test
public void testWhenAckMoreThanCk() throws Throwable {
brokerConfig.setEnableSkipLongAwaitingAck(true);
Expand Down
Loading