This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new b764205990 [ISSUE #11272] Defer LMQ arrival notification to
post-commit (#11273)
b764205990 is described below
commit b76420599098c6f53bdf80544f8a1b7ffe678fa9
Author: imzs <[email protected]>
AuthorDate: Mon Oct 12 13:36:26 2026 +0800
[ISSUE #11272] Defer LMQ arrival notification to post-commit (#11273)
---
.../apache/rocketmq/store/DefaultMessageStore.java | 23 +++-
.../rocketmq/store/DefaultMessageStoreTest.java | 126 +++++++++++++++++----
2 files changed, 126 insertions(+), 23 deletions(-)
diff --git
a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
index e2d95b963d..3bd6a95a2b 100644
--- a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
+++ b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
@@ -112,6 +112,7 @@ import
org.apache.rocketmq.store.queue.ConsumeQueueInterface;
import org.apache.rocketmq.store.queue.ConsumeQueueStore;
import org.apache.rocketmq.store.queue.ConsumeQueueStoreInterface;
import org.apache.rocketmq.store.queue.CqUnit;
+import org.apache.rocketmq.store.queue.MultiDispatchUtils;
import org.apache.rocketmq.store.queue.ReferredIterator;
import org.apache.rocketmq.store.rocksdb.MessageRocksDBStorage;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
@@ -2642,13 +2643,24 @@ public class DefaultMessageStore implements
MessageStore {
@Override
public void notifyMessageArriveIfNecessary(DispatchRequest
dispatchRequest) {
+ // keeps original parameters and behavior by always passing true here
+ notifyMessageArrive(dispatchRequest, true);
+ }
+
+ /**
+ * Notify message arrival, optionally including the multi-queue (LMQ) part.
+ * @param notifyMultiQueue whether to also notify the multi-queue arrival
+ */
+ private void notifyMessageArrive(DispatchRequest dispatchRequest, boolean
notifyMultiQueue) {
if (DefaultMessageStore.this.brokerConfig.isLongPollingEnable()
&& DefaultMessageStore.this.messageArrivingListener != null) {
DefaultMessageStore.this.messageArrivingListener.arriving(dispatchRequest.getTopic(),
dispatchRequest.getQueueId(),
dispatchRequest.getConsumeQueueOffset() + 1,
dispatchRequest.getTagsCode(),
dispatchRequest.getStoreTimestamp(),
dispatchRequest.getBitMap(),
dispatchRequest.getPropertiesMap());
-
DefaultMessageStore.this.reputMessageService.notifyMessageArrive4MultiQueue(dispatchRequest);
+ if (notifyMultiQueue) {
+
DefaultMessageStore.this.reputMessageService.notifyMessageArrive4MultiQueue(dispatchRequest);
+ }
}
}
@@ -2745,7 +2757,10 @@ public class DefaultMessageStore implements MessageStore
{
DefaultMessageStore.this.doDispatch(dispatchRequest);
if (isNotifyMessageArriveWhenReput()) {
-
notifyMessageArriveIfNecessary(dispatchRequest);
+ // only notify multi-queue when we do real
LMQ dispatch action.
+ boolean notifyMultiQueue =
MultiDispatchUtils.checkMultiDispatchQueue(
+
DefaultMessageStore.this.messageStoreConfig, dispatchRequest);
+ notifyMessageArrive(dispatchRequest,
notifyMultiQueue);
}
this.reputFromOffset += size;
@@ -2929,7 +2944,9 @@ public class DefaultMessageStore implements MessageStore {
for (DispatchRequest dispatchRequest : dispatchRequests) {
DefaultMessageStore.this.doDispatch(dispatchRequest);
// wake up long-polling
-
DefaultMessageStore.this.notifyMessageArriveIfNecessary(dispatchRequest);
+ boolean notifyMultiQueue =
MultiDispatchUtils.checkMultiDispatchQueue(
+ DefaultMessageStore.this.messageStoreConfig,
dispatchRequest);
+ notifyMessageArrive(dispatchRequest, notifyMultiQueue);
if
(!DefaultMessageStore.this.getMessageStoreConfig().isDuplicationEnable() &&
DefaultMessageStore.this.getMessageStoreConfig().getBrokerRole() ==
BrokerRole.SLAVE) {
diff --git
a/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java
b/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java
index 05e186537b..91289af9ef 100644
--- a/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java
@@ -20,13 +20,17 @@ package org.apache.rocketmq.store;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.any;
import static org.mockito.Mockito.anyLong;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.same;
import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -59,8 +63,12 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.UtilAll;
+import org.apache.rocketmq.common.attribute.TopicMessageType;
+import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageBatch;
+import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageExtBatch;
@@ -83,6 +91,8 @@ import org.mockito.junit.MockitoJUnitRunner;
@RunWith(MockitoJUnitRunner.class)
public class DefaultMessageStoreTest {
+ private static final String DEFAULT_STORE_PATH_ROOT_DIR = new
MessageStoreConfig().getStorePathRootDir();
+
private final String storeMessage = "Once, there was a chance for me!";
private final String messageTopic = "FooBar";
private int queueTotal = 100;
@@ -131,35 +141,39 @@ public class DefaultMessageStoreTest {
@After
public void destroy() {
- messageStore.shutdown();
- messageStore.destroy();
+ shutdownAndDeleteStore(messageStore);
+ }
- MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
- File file = new File(messageStoreConfig.getStorePathRootDir());
- UtilAll.deleteFile(file);
+ private void shutdownAndDeleteStore(MessageStore store) {
+ store.shutdown();
+ store.destroy();
+ UtilAll.deleteFile(new
File(store.getMessageStoreConfig().getStorePathRootDir()));
}
private MessageStore buildMessageStore() throws Exception {
- return buildMessageStore(null);
+ return buildMessageStore(null, null);
}
private MessageStore buildMessageStore(String storePathRootDir) throws
Exception {
- MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
- messageStoreConfig.setMappedFileSizeCommitLog(1024 * 1024 * 10);
- messageStoreConfig.setMappedFileSizeConsumeQueue(1024 * 1024 * 10);
- messageStoreConfig.setMaxHashSlotNum(10000);
- messageStoreConfig.setMaxIndexNum(100 * 100);
- messageStoreConfig.setFlushDiskType(FlushDiskType.SYNC_FLUSH);
- messageStoreConfig.setFlushIntervalConsumeQueue(1);
- messageStoreConfig.setHaListenPort(0);
- if (Strings.isNullOrEmpty(storePathRootDir)) {
- UUID uuid = UUID.randomUUID();
- storePathRootDir = System.getProperty("java.io.tmpdir") +
File.separator + "store-" + uuid.toString();
+ MessageStoreConfig storeConfig = new MessageStoreConfig();
+ storeConfig.setStorePathRootDir(storePathRootDir);
+ return buildMessageStore(storeConfig, null);
+ }
+
+ private MessageStore buildMessageStore(MessageStoreConfig storeConfig,
MessageArrivingListener listener) throws Exception {
+ if (storeConfig == null) {
+ storeConfig = new MessageStoreConfig();
}
- messageStoreConfig.setStorePathRootDir(storePathRootDir);
- return new DefaultMessageStore(messageStoreConfig,
+ if (Strings.isNullOrEmpty(storeConfig.getStorePathRootDir())
+ ||
DEFAULT_STORE_PATH_ROOT_DIR.equals(storeConfig.getStorePathRootDir())) {
+
storeConfig.setStorePathRootDir(System.getProperty("java.io.tmpdir") +
File.separator + "store-" + UUID.randomUUID());
+ }
+ storeConfig.setHaListenPort(0);
+ storeConfig.setFlushDiskType(FlushDiskType.SYNC_FLUSH);
+ storeConfig.setFlushIntervalConsumeQueue(1);
+ return new DefaultMessageStore(storeConfig,
new BrokerStatsManager("simpleTest", true),
- new MyMessageArrivingListener(),
+ listener != null ? listener : new MyMessageArrivingListener(),
new BrokerConfig(), new ConcurrentHashMap<>());
}
@@ -1118,6 +1132,78 @@ public class DefaultMessageStoreTest {
store.destroy();
}
+ @Test
+ public void testReputNotify_notifyMultiQueue() throws Exception {
+ String lmq = "%LMQ%notify";
+ messageBody = storeMessage.getBytes();
+ MessageArrivingListener listener = mock(MessageArrivingListener.class);
+
+ MessageStoreConfig storeConfig = new MessageStoreConfig();
+ storeConfig.setEnableLmq(true);
+ storeConfig.setEnableMultiDispatch(true);
+
+ DefaultMessageStore store = (DefaultMessageStore)
buildMessageStore(storeConfig, listener);
+ assertTrue(store.load());
+ store.start();
+ try {
+ store.putMessage(buildLmqMessage(lmq));
+ StoreTestUtil.waitCommitLogReput(store);
+ assertThat(store.getConsumeQueueTable().get(lmq)).isNotNull();
+ verify(listener).arriving(eq(lmq), eq(MixAll.LMQ_QUEUE_ID),
eq(1L), anyLong(), anyLong(), any(), any());
+ } finally {
+ shutdownAndDeleteStore(store);
+ }
+ }
+
+ @Test
+ public void testReputNotify_deferNotifyToGroupCommitForLite() throws
Exception {
+ String lmq = "%LMQ%notify";
+ messageBody = storeMessage.getBytes();
+ MessageArrivingListener listener = mock(MessageArrivingListener.class);
+
+ MessageStoreConfig storeConfig = new MessageStoreConfig();
+ storeConfig.setEnableLmq(true);
+ storeConfig.setEnableMultiDispatch(true);
+ // Combine store where the LMQ consume queue is committed
asynchronously by RocksGroupCommitService
+ storeConfig.setRocksdbCQDoubleWriteEnable(true);
+ storeConfig.setRocksdbCQSelectiveDoubleWriteEnable(true);
+ storeConfig.setCombineCQUseRocksdbForLmq(true);
+
+ DefaultMessageStore store = (DefaultMessageStore)
buildMessageStore(storeConfig, listener);
+ // selective double-write only routes LITE topics to rocksdb
+ TopicConfig liteTopic = new TopicConfig(messageTopic);
+ liteTopic.setTopicMessageType(TopicMessageType.LITE);
+ store.getTopicConfigs().put(messageTopic, liteTopic);
+ assertTrue(store.load());
+ store.start();
+ try {
+ store.putMessage(buildLmqMessage(lmq));
+ StoreTestUtil.waitCommitLogReput(store);
+
+ verify(listener, timeout(5000).times(1))
+ .arriving(eq(lmq), eq(MixAll.LMQ_QUEUE_ID), eq(1L), anyLong(),
anyLong(), any(), any());
+ verify(listener, times(2)).arriving(eq(messageTopic), eq(0),
eq(1L), anyLong(), anyLong(), any(), any());
+ } finally {
+ shutdownAndDeleteStore(store);
+ }
+ }
+
+ private MessageExtBrokerInner buildLmqMessage(String lmq) {
+ MessageExtBrokerInner msg = new MessageExtBrokerInner();
+ msg.setTopic(messageTopic);
+ msg.setTags("TAG1");
+ msg.setKeys("Hello");
+ msg.setBody(messageBody);
+ msg.setQueueId(0);
+ msg.setSysFlag(0);
+ msg.setBornTimestamp(System.currentTimeMillis());
+ msg.setStoreHost(storeHost);
+ msg.setBornHost(bornHost);
+ MessageAccessor.putProperty(msg,
MessageConst.PROPERTY_INNER_MULTI_DISPATCH, lmq);
+
msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));
+ return msg;
+ }
+
private class MyMessageArrivingListener implements MessageArrivingListener
{
@Override
public void arriving(String topic, int queueId, long logicOffset, long
tagsCode, long msgStoreTime,
