[ISUUE #8762] Fix dLedger lock granularity issue causing LMQ message loss

store/pom.xml

@@ -32,6 +32,10 @@
     </properties>
 
     <dependencies>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>rocketmq-common</artifactId>
+        </dependency>
         <dependency>
             <groupId>io.openmessaging.storage</groupId>
             <artifactId>dledger</artifactId>

store/src/main/java/org/apache/rocketmq/store/dledger/DLedgerCommitLog.java

@@ -41,6 +41,7 @@
 import org.apache.rocketmq.store.StoreStatsService;
 import org.apache.rocketmq.store.config.MessageStoreConfig;
 import org.apache.rocketmq.store.logfile.MappedFile;
+import org.apache.rocketmq.store.queue.MultiDispatchUtils;
 import org.rocksdb.RocksDBException;
 
 import io.openmessaging.storage.dledger.AppendFuture;
@@ -569,6 +570,7 @@ public CompletableFuture<PutMessageResult> asyncPutMessage(MessageExtBrokerInner
         EncodeResult encodeResult;
 
         String topicQueueKey = msg.getTopic() + "-" + msg.getQueueId();
+        final boolean isMultiDispatchMsg = CommitLog.isMultiDispatchMsg(this.defaultMessageStore.getMessageStoreConfig(), msg);
         topicQueueLock.lock(topicQueueKey);
         try {
             defaultMessageStore.assignOffset(msg);
@@ -580,6 +582,18 @@ public CompletableFuture<PutMessageResult> asyncPutMessage(MessageExtBrokerInner
             putMessageLock.lock(); //spin or ReentrantLock ,depending on store config
             long elapsedTimeInLock;
             long queueOffset;
+            if (isMultiDispatchMsg) {
+                AppendMessageResult appendMessageResult = MultiDispatchUtils.handlePropertiesForLmqMsg(
+                        encodeResult.data, msg, this.multiDispatch, this.getMessageStore().getMessageStoreConfig());
+                if (appendMessageResult != null) {
+                    PutMessageStatus putMessageStatus = PutMessageStatus.MESSAGE_ILLEGAL;
+                    return CompletableFuture.completedFuture(new PutMessageResult(putMessageStatus, appendMessageResult));
+                }
+            }
+            final int msgLen = encodeResult.data.getInt(0);
+            encodeResult.data.position(0);
+            encodeResult.data.limit(msgLen);
+
             try {
                 beginTimeInDledgerLock = this.defaultMessageStore.getSystemClock().now();
                 queueOffset = getQueueOffsetByKey(msg, tranType);
@@ -599,6 +613,9 @@ public CompletableFuture<PutMessageResult> asyncPutMessage(MessageExtBrokerInner
 
                 String msgId = MessageDecoder.createMessageId(buffer, msg.getStoreHostBytes(), wroteOffset);
                 elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginTimeInDledgerLock;
+                if (isMultiDispatchMsg) {
+                    this.multiDispatch.updateMultiQueueOffset(msg);
+                }
                 appendResult = new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, encodeResult.getData().length, msgId, System.currentTimeMillis(), queueOffset, elapsedTimeInLock);
             } finally {
                 beginTimeInDledgerLock = 0;

store/src/main/java/org/apache/rocketmq/store/queue/MultiDispatchUtils.java

@@ -16,16 +16,27 @@
  */
 package org.apache.rocketmq.store.queue;
 
+import java.nio.ByteBuffer;
 import java.util.Map;
 
 import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.common.message.MessageConst;
+import org.apache.rocketmq.common.message.MessageDecoder;
+import org.apache.rocketmq.common.message.MessageExtBrokerInner;
 import org.apache.rocketmq.common.topic.TopicValidator;
+import org.apache.rocketmq.logging.org.slf4j.Logger;
+import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
+import org.apache.rocketmq.store.AppendMessageResult;
+import org.apache.rocketmq.store.AppendMessageStatus;
+import org.apache.rocketmq.store.CommitLog;
 import org.apache.rocketmq.store.DispatchRequest;
+import org.apache.rocketmq.store.MultiDispatch;
 import org.apache.rocketmq.store.config.MessageStoreConfig;
 
 public class MultiDispatchUtils {
+    private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
 
     public static String lmqQueueKey(String queueName) {
         StringBuilder keyBuilder = new StringBuilder();
@@ -58,4 +69,62 @@ public static boolean checkMultiDispatchQueue(MessageStoreConfig messageStoreCon
         }
         return true;
     }
+
+    public static AppendMessageResult handlePropertiesForLmqMsg(
+            ByteBuffer preEncodeBuffer, final MessageExtBrokerInner msgInner, MultiDispatch multiDispatch, MessageStoreConfig messageStoreConfig) {
+        if (msgInner.isEncodeCompleted()) {
+            return null;
+        }
+
+        int crc32ReservedLength = messageStoreConfig.isEnabledAppendPropCRC() ? CommitLog.CRC32_RESERVED_LEN : 0;
+
+        multiDispatch.wrapMultiDispatch(msgInner);
+
+        msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
+
+        final byte[] propertiesData =
+                msgInner.getPropertiesString() == null ? null : msgInner.getPropertiesString().getBytes(MessageDecoder.CHARSET_UTF8);
+
+        boolean needAppendLastPropertySeparator = messageStoreConfig.isEnabledAppendPropCRC() && propertiesData != null && propertiesData.length > 0
+                && propertiesData[propertiesData.length - 1] != MessageDecoder.PROPERTY_SEPARATOR;
+
+        final int propertiesLength =
+            (propertiesData == null ? 0 : propertiesData.length) + (needAppendLastPropertySeparator ? 1 : 0) + crc32ReservedLength;
+
+        if (propertiesLength > Short.MAX_VALUE) {
+            log.warn("putMessage message properties length too long. length={}", propertiesData.length);
+            return new AppendMessageResult(AppendMessageStatus.PROPERTIES_SIZE_EXCEEDED);
+        }
+
+        int msgLenWithoutProperties = preEncodeBuffer.getInt(0);
+
+        int msgLen = msgLenWithoutProperties + 2 + propertiesLength;
+
+        // Exceeds the maximum message
+        if (msgLen > messageStoreConfig.getMaxMessageSize()) {
+            log.warn("message size exceeded, msg total size: " + msgLen + ", maxMessageSize: " + messageStoreConfig.getMaxMessageSize());
+            return new AppendMessageResult(AppendMessageStatus.MESSAGE_SIZE_EXCEEDED);
+        }
+
+        // Back filling total message length
+        preEncodeBuffer.putInt(0, msgLen);
+        // Modify position to msgLenWithoutProperties
+        preEncodeBuffer.position(msgLenWithoutProperties);
+
+        preEncodeBuffer.putShort((short) propertiesLength);
+
+        if (propertiesLength > crc32ReservedLength) {
+            preEncodeBuffer.put(propertiesData);
+        }
+
+        if (needAppendLastPropertySeparator) {
+            preEncodeBuffer.put((byte) MessageDecoder.PROPERTY_SEPARATOR);
+        }
+        // 18 CRC32
+        preEncodeBuffer.position(preEncodeBuffer.position() + crc32ReservedLength);
+
+        msgInner.setEncodeCompleted(true);
+
+        return null;
+    }
 }