Skip to content

[ISSUE #10813] Reduce temporary allocations in the LMQ append path #10814

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

Merged
fuyou001 merged 1 commit into from
Aug 6, 2026
Merged

[ISSUE #10813] Reduce temporary allocations in the LMQ append path #10814

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
32 changes: 20 additions & 12 deletions store/src/main/java/org/apache/rocketmq/store/CommitLog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1939,17 +1939,7 @@ public AppendMessageResult handlePropertiesForLmqMsg(ByteBuffer preEncodeBuffer,
return null;
}

try {
LmqDispatch.wrapLmqDispatch(defaultMessageStore, msgInner);
} catch (ConsumeQueueException e) {
if (e.getCause() instanceof RocksDBException) {
log.error("Failed to wrap multi-dispatch", e);
return new AppendMessageResult(AppendMessageStatus.ROCKSDB_ERROR);
}
log.error("Failed to wrap multi-dispatch", e);
return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
}

LmqDispatch.reinsertWaitStorePropertyForLegacySerialization(msgInner);
msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));

final byte[] propertiesData =
Expand Down Expand Up @@ -2003,7 +1993,20 @@ public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer

ByteBuffer preEncodeBuffer = msgInner.getEncodedBuff();
boolean isMultiDispatchMsg = messageStoreConfig.isEnableLmq() && msgInner.needDispatchLMQ();
String[] lmqQueueNames = null;
if (isMultiDispatchMsg) {
if (!msgInner.isEncodeCompleted()) {
try {
lmqQueueNames = LmqDispatch.prepareLmqDispatch(defaultMessageStore, msgInner);
} catch (ConsumeQueueException e) {
if (e.getCause() instanceof RocksDBException) {
log.error("Failed to wrap multi-dispatch", e);
return new AppendMessageResult(AppendMessageStatus.ROCKSDB_ERROR);
}
log.error("Failed to wrap multi-dispatch", e);
return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
}
}
AppendMessageResult appendMessageResult = handlePropertiesForLmqMsg(preEncodeBuffer, msgInner);
if (appendMessageResult != null) {
return appendMessageResult;
Expand Down Expand Up @@ -2099,7 +2102,12 @@ public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer

if (isMultiDispatchMsg) {
try {
LmqDispatch.updateLmqOffsets(defaultMessageStore, msgInner);
if (lmqQueueNames == null) {
// The encoded message may be retried after reaching the end of a mapped file.
LmqDispatch.updateLmqOffsets(defaultMessageStore, msgInner);
} else {
LmqDispatch.updateLmqOffsets(defaultMessageStore, lmqQueueNames);
}
} catch (ConsumeQueueException e) {
// Increase in-memory max offset of the queue should not fail.
return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
Expand Down
55 changes: 41 additions & 14 deletions store/src/main/java/org/apache/rocketmq/store/LmqDispatch.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@
*/
package org.apache.rocketmq.store;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageConst;
Expand All @@ -28,25 +27,53 @@ public class LmqDispatch {

public static void wrapLmqDispatch(MessageStore messageStore, final MessageExtBrokerInner msg)
throws ConsumeQueueException {
String lmqNames = msg.getProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH);
String[] queueNames = lmqNames.split(MixAll.LMQ_DISPATCH_SEPARATOR);
Long[] queueOffsets = new Long[queueNames.length];
if (messageStore.getMessageStoreConfig().isEnableLmq()) {
for (int i = 0; i < queueNames.length; i++) {
if (MixAll.isLmq(queueNames[i])) {
queueOffsets[i] = messageStore.getQueueStore().getLmqQueueOffset(queueNames[i], MixAll.LMQ_QUEUE_ID);
}
populateLmqOffsets(messageStore, msg);
msg.removeWaitStorePropertyString();
}

static String[] prepareLmqDispatch(MessageStore messageStore, final MessageExtBrokerInner msg)
throws ConsumeQueueException {
return populateLmqOffsets(messageStore, msg);
}

static void reinsertWaitStorePropertyForLegacySerialization(final MessageExtBrokerInner msg) {
// Reproduce the legacy remove/reinsert mutation without the discarded serialization.
if (msg.getProperties().containsKey(MessageConst.PROPERTY_WAIT_STORE_MSG_OK)) {
String waitStoreMsgOKValue = msg.getProperties().remove(MessageConst.PROPERTY_WAIT_STORE_MSG_OK);
msg.getProperties().put(MessageConst.PROPERTY_WAIT_STORE_MSG_OK, waitStoreMsgOKValue);
}
}

private static String[] populateLmqOffsets(MessageStore messageStore, final MessageExtBrokerInner msg)
throws ConsumeQueueException {
String[] queueNames = parseLmqQueueNames(msg);
StringBuilder queueOffsets = new StringBuilder();
boolean enableLmq = messageStore.getMessageStoreConfig().isEnableLmq();
for (int i = 0; i < queueNames.length; i++) {
if (i > 0) {
queueOffsets.append(MixAll.LMQ_DISPATCH_SEPARATOR);
}
if (enableLmq && MixAll.isLmq(queueNames[i])) {
queueOffsets.append(messageStore.getQueueStore().getLmqQueueOffset(queueNames[i],
MixAll.LMQ_QUEUE_ID));
}
}
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET,
StringUtils.join(queueOffsets, MixAll.LMQ_DISPATCH_SEPARATOR));
msg.removeWaitStorePropertyString();
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET, queueOffsets.toString());
return queueNames;
}

private static String[] parseLmqQueueNames(final MessageExtBrokerInner msg) {
String lmqNames = msg.getProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH);
return lmqNames.split(MixAll.LMQ_DISPATCH_SEPARATOR);
}

public static void updateLmqOffsets(MessageStore messageStore, final MessageExtBrokerInner msgInner)
throws ConsumeQueueException {
String lmqNames = msgInner.getProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH);
String[] queueNames = lmqNames.split(MixAll.LMQ_DISPATCH_SEPARATOR);
updateLmqOffsets(messageStore, parseLmqQueueNames(msgInner));
}

static void updateLmqOffsets(MessageStore messageStore, String[] queueNames)
throws ConsumeQueueException {
for (String queueName : queueNames) {
if (messageStore.getMessageStoreConfig().isEnableLmq() && MixAll.isLmq(queueName)) {
messageStore.getQueueStore().increaseLmqOffset(queueName, MixAll.LMQ_QUEUE_ID, VALUE_OF_EACH_INCREMENT);
Expand Down
Loading
Loading