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 80e1ae5577 [ISSUE #11163] Fix lite topic prefix index not maintained
on first message (#11164)
80e1ae5577 is described below
commit 80e1ae55773c2330d2005b86f020ed028f649e94
Author: Quan <[email protected]>
AuthorDate: Tue Sep 15 10:03:56 2026 +0800
[ISSUE #11163] Fix lite topic prefix index not maintained on first message
(#11164)
- Correct the off-by-one offset check in LiteEventDispatcher.dispatch
(offset == 0 -> == 1) so onLmqCreate fires on the first arriving message
- Restore store-dependent lifecycle tests to the concrete impl test
classes, driven by real putMessage through the notify path
- Add MessageArrivingListener overloads to LiteTestUtil to wire the real
dispatch path in tests
---
.../rocketmq/broker/lite/LiteEventDispatcher.java | 7 +-
.../lite/AbstractLiteLifecycleManagerTest.java | 108 ---------------------
.../broker/lite/LiteLifecycleManagerTest.java | 93 +++++++++++++++++-
.../apache/rocketmq/broker/lite/LiteTestUtil.java | 19 +++-
.../lite/RocksDBLiteLifecycleManagerTest.java | 107 ++++++++++++++++++--
5 files changed, 209 insertions(+), 125 deletions(-)
diff --git
a/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteEventDispatcher.java
b/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteEventDispatcher.java
index 7018a46344..d7c064d242 100644
---
a/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteEventDispatcher.java
+++
b/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteEventDispatcher.java
@@ -90,9 +90,10 @@ public class LiteEventDispatcher extends ServiceThread {
if (queueId != 0 || !LiteUtil.isLiteTopicQueue(lmqName)) {
return;
}
- // Maintain prefix index only on the lmq's first message; pre-existing
lmqs are
- // populated once at startup during init().
- if (offset == 0) {
+ // Maintain prefix index on the lmq's first message: the arriving
notification carries
+ // logicOffset = queueOffset + 1, so the first message shows offset ==
1. Pre-existing lmqs
+ // are populated once at startup during init().
+ if (offset == 1) {
liteLifecycleManager.onLmqCreate(lmqName);
}
doDispatch(group, lmqName, null);
diff --git
a/broker/src/test/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManagerTest.java
b/broker/src/test/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManagerTest.java
index b3eb91c373..4a15311a79 100644
---
a/broker/src/test/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManagerTest.java
+++
b/broker/src/test/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManagerTest.java
@@ -18,9 +18,7 @@
package org.apache.rocketmq.broker.lite;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
-import java.util.HashSet;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -47,7 +45,6 @@ import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
-import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import org.mockito.junit.MockitoJUnitRunner;
@@ -59,7 +56,6 @@ import static org.mockito.Mockito.when;
import static org.mockito.Mockito.anyInt;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.never;
import static org.mockito.Mockito.atLeastOnce;
@RunWith(MockitoJUnitRunner.class)
@@ -112,7 +108,6 @@ public class AbstractLiteLifecycleManagerTest {
topicConfig.setLiteTopicExpiration(1);
topicConfigTable.put(PARENT_TOPIC, topicConfig);
when(topicConfigManager.getTopicConfigTable()).thenReturn(topicConfigTable);
-
when(topicConfigManager.selectTopicConfig(PARENT_TOPIC)).thenReturn(topicConfig);
groupConfig.setGroupName(GROUP);
groupConfig.setLiteBindTopic(PARENT_TOPIC);
@@ -158,28 +153,6 @@ public class AbstractLiteLifecycleManagerTest {
Assert.assertFalse(lifecycleManager.isLmqExist("whatever"));
}
- @Test
- public void testGetLiteTopicCount() {
- Assert.assertEquals(1,
lifecycleManager.getLiteTopicCount(PARENT_TOPIC));
- Assert.assertEquals(0, lifecycleManager.getLiteTopicCount("whatever"));
-
- // parentTopic1: 2 liteTopics, parentTopic2: 3 liteTopics
- String parent1 = "parentTopic1";
- String parent2 = "parentTopic2";
- registerLiteTopicConfig(parent1);
- registerLiteTopicConfig(parent2);
- lifecycleManager.lmqPrefixIndex.add(LiteUtil.toLmqName(parent1,
"sub1"));
- lifecycleManager.lmqPrefixIndex.add(LiteUtil.toLmqName(parent1,
"sub2"));
- lifecycleManager.lmqPrefixIndex.add(LiteUtil.toLmqName(parent2,
"sub1"));
- lifecycleManager.lmqPrefixIndex.add(LiteUtil.toLmqName(parent2,
"sub2"));
- lifecycleManager.lmqPrefixIndex.add(LiteUtil.toLmqName(parent2,
"sub3"));
-
- Assert.assertEquals(2, lifecycleManager.getLiteTopicCount(parent1));
- Assert.assertEquals(3, lifecycleManager.getLiteTopicCount(parent2));
- // PARENT_TOPIC count unchanged
- Assert.assertEquals(1,
lifecycleManager.getLiteTopicCount(PARENT_TOPIC));
- }
-
@Test
public void testIsLiteTopicExpired() {
// not lite topic queue
@@ -268,79 +241,6 @@ public class AbstractLiteLifecycleManagerTest {
verify(liteSubscriptionRegistry).cleanSubscription(EXIST_LMQ_NAME,
false);
}
- @Test
- public void testCleanByParentTopic() {
- String lmq1 = LiteUtil.toLmqName(PARENT_TOPIC, "sub1");
- String lmq2 = LiteUtil.toLmqName(PARENT_TOPIC, "sub2");
- String lmq3 = LiteUtil.toLmqName(PARENT_TOPIC, "sub3");
-
- String otherLmq1 = LiteUtil.toLmqName("otherParentTopic", "sub1");
- String otherLmq2 = LiteUtil.toLmqName("otherParentTopic", "sub2");
-
- // multiple LMQs: deleteLmq called only for LMQs under PARENT_TOPIC
- lifecycleManager.lmqPrefixIndex.remove(EXIST_LMQ_NAME);
- lifecycleManager.lmqPrefixIndex.add(lmq1);
- lifecycleManager.lmqPrefixIndex.add(lmq2);
- lifecycleManager.lmqPrefixIndex.add(lmq3);
- lifecycleManager.lmqPrefixIndex.add(otherLmq1);
- lifecycleManager.lmqPrefixIndex.add(otherLmq2);
-
- ArgumentCaptor<String> parentCaptor =
ArgumentCaptor.forClass(String.class);
- ArgumentCaptor<String> lmqCaptor =
ArgumentCaptor.forClass(String.class);
- lifecycleManager.cleanByParentTopic(PARENT_TOPIC);
- verify(lifecycleManager, times(3)).deleteLmq(parentCaptor.capture(),
lmqCaptor.capture());
-
Assert.assertTrue(parentCaptor.getAllValues().stream().allMatch(PARENT_TOPIC::equals));
- Assert.assertEquals(new HashSet<>(Arrays.asList(lmq1, lmq2, lmq3)),
new HashSet<>(lmqCaptor.getAllValues()));
-
- // other parent's LMQs remain untouched
- List<String> otherResult =
lifecycleManager.collectByParentTopic("otherParentTopic");
- Assert.assertEquals(new HashSet<>(Arrays.asList(otherLmq1,
otherLmq2)), new HashSet<>(otherResult));
-
- // zero LMQs: deleteLmq not called
- Mockito.clearInvocations(lifecycleManager);
- lifecycleManager.cleanByParentTopic(PARENT_TOPIC);
- verify(lifecycleManager, never()).deleteLmq(anyString(), anyString());
-
- // guard: non-lite topic and null both return early
- Mockito.clearInvocations(lifecycleManager);
- lifecycleManager.lmqPrefixIndex.add(EXIST_LMQ_NAME);
- lifecycleManager.cleanByParentTopic("nonExistentTopic");
- verify(lifecycleManager, never()).deleteLmq(anyString(), anyString());
- lifecycleManager.cleanByParentTopic(null);
- verify(lifecycleManager, never()).deleteLmq(anyString(), anyString());
- }
-
- @Test
- public void testCollectByParentTopic() {
- String lmq1 = LiteUtil.toLmqName(PARENT_TOPIC, "sub1");
- String lmq2 = LiteUtil.toLmqName(PARENT_TOPIC, "sub2");
- String lmq3 = LiteUtil.toLmqName(PARENT_TOPIC, "sub3");
-
- String otherLmq1 = LiteUtil.toLmqName("otherParentTopic", "sub1");
- String otherLmq2 = LiteUtil.toLmqName("otherParentTopic", "sub2");
-
- lifecycleManager.lmqPrefixIndex.remove(EXIST_LMQ_NAME);
- lifecycleManager.lmqPrefixIndex.add(lmq1);
- lifecycleManager.lmqPrefixIndex.add(lmq2);
- lifecycleManager.lmqPrefixIndex.add(lmq3);
- lifecycleManager.lmqPrefixIndex.add(otherLmq1);
- lifecycleManager.lmqPrefixIndex.add(otherLmq2);
-
- // multiple LMQs: returns only those under PARENT_TOPIC, excluding
other parent's
- List<String> result =
lifecycleManager.collectByParentTopic(PARENT_TOPIC);
- Assert.assertEquals(new HashSet<>(Arrays.asList(lmq1, lmq2, lmq3)),
new HashSet<>(result));
-
- // no LMQs under parent: returns empty list
- result = lifecycleManager.collectByParentTopic("nonExistentTopic");
- Assert.assertTrue(result.isEmpty());
-
- // guard: null and empty both return empty list
- result = lifecycleManager.collectByParentTopic(null);
- Assert.assertTrue(result.isEmpty());
- result = lifecycleManager.collectByParentTopic("");
- Assert.assertTrue(result.isEmpty());
- }
-
@Test
public void testRun() throws InterruptedException {
brokerConfig.setLiteTtlCheckInterval(100L);
@@ -357,14 +257,6 @@ public class AbstractLiteLifecycleManagerTest {
verify(liteSubscriptionRegistry,
atLeastOnce()).cleanSubscription(EXIST_LMQ_NAME, false);
}
- private void registerLiteTopicConfig(String parentTopic) {
- TopicConfig config = new TopicConfig(parentTopic, 1, 1);
- config.getAttributes().put(
- TopicAttributes.TOPIC_MESSAGE_TYPE_ATTRIBUTE.getName(),
TopicMessageType.LITE.getValue());
- topicConfigTable.put(parentTopic, config);
-
when(topicConfigManager.selectTopicConfig(parentTopic)).thenReturn(config);
- }
-
private static class TestLiteLifecycleManager extends
AbstractLiteLifecycleManager {
public TestLiteLifecycleManager(BrokerController brokerController,
LiteSharding liteSharding) {
diff --git
a/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteLifecycleManagerTest.java
b/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteLifecycleManagerTest.java
index e936fc1805..cf5d872d88 100644
---
a/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteLifecycleManagerTest.java
+++
b/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteLifecycleManagerTest.java
@@ -18,13 +18,20 @@
package org.apache.rocketmq.broker.lite;
import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.broker.longpolling.NotifyMessageArrivingListener;
+import org.apache.rocketmq.broker.longpolling.PullRequestHoldService;
import org.apache.rocketmq.broker.offset.ConsumerOffsetManager;
+import org.apache.rocketmq.broker.processor.NotificationProcessor;
+import org.apache.rocketmq.broker.processor.PopMessageProcessor;
import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
import org.apache.rocketmq.broker.topic.TopicConfigManager;
import org.apache.rocketmq.common.BrokerConfig;
+import org.apache.rocketmq.common.TopicAttributes;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.UtilAll;
+import org.apache.rocketmq.common.attribute.TopicMessageType;
import org.apache.rocketmq.common.lite.LiteUtil;
+import org.apache.rocketmq.store.MessageArrivingListener;
import org.apache.rocketmq.store.MessageStore;
import org.junit.AfterClass;
import org.junit.Assert;
@@ -36,6 +43,7 @@ import org.mockito.Mockito;
import org.mockito.junit.MockitoJUnitRunner;
import java.io.File;
+import java.util.Collections;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
@@ -49,6 +57,7 @@ import static org.awaitility.Awaitility.await;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.nullable;
import static org.mockito.Mockito.when;
@RunWith(MockitoJUnitRunner.Silent.class)
@@ -59,6 +68,7 @@ public class LiteLifecycleManagerTest {
private static String storePathRootDir;
private static MessageStore messageStore;
private static LiteLifecycleManager liteLifecycleManager;
+ private static LiteEventDispatcher liteEventDispatcher;
private static TopicConfig mockTopicConfig = new TopicConfig();
@BeforeClass
@@ -66,10 +76,6 @@ public class LiteLifecycleManagerTest {
storePathRootDir = System.getProperty("java.io.tmpdir") +
File.separator + "store-lifecycleTest";
UtilAll.deleteFile(new File(storePathRootDir));
- messageStore = LiteTestUtil.buildMessageStore(storePathRootDir,
BROKER_CONFIG, TOPIC_CONFIG_TABLE, false);
- messageStore.load();
- messageStore.start();
-
BrokerController brokerController =
Mockito.mock(BrokerController.class);
LiteSharding liteSharding = Mockito.mock(LiteSharding.class);
TopicConfigManager topicConfigManager =
Mockito.mock(TopicConfigManager.class);
@@ -77,9 +83,11 @@ public class LiteLifecycleManagerTest {
LiteSubscriptionRegistry liteSubscriptionRegistry =
Mockito.mock(LiteSubscriptionRegistry.class);
ConsumerOffsetManager consumerOffsetManager =
Mockito.mock(ConsumerOffsetManager.class);
when(consumerOffsetManager.getPullOffsetTable()).thenReturn(new
ConcurrentHashMap<>());
+ // dispatch() fans out to subscribers right after maintaining the
prefix index; no subscriber here
+
when(liteSubscriptionRegistry.getAllSubscribers(nullable(String.class),
anyString()))
+ .thenReturn(Collections.emptyMap());
when(brokerController.getBrokerConfig()).thenReturn(BROKER_CONFIG);
- when(brokerController.getMessageStore()).thenReturn(messageStore);
when(brokerController.getTopicConfigManager()).thenReturn(topicConfigManager);
when(brokerController.getSubscriptionGroupManager()).thenReturn(subscriptionGroupManager);
when(brokerController.getLiteSubscriptionRegistry()).thenReturn(liteSubscriptionRegistry);
@@ -90,6 +98,20 @@ public class LiteLifecycleManagerTest {
LiteLifecycleManager testObject = new
LiteLifecycleManager(brokerController, liteSharding);
liteLifecycleManager = Mockito.spy(testObject);
+
+ // Wire the real notify path so putMessage drives the prefix index:
+ // putMessage -> reput -> NotifyMessageArrivingListener.arriving
+ // -> LiteEventDispatcher.dispatch -> onLmqCreate -> lmqPrefixIndex
+ liteEventDispatcher = new LiteEventDispatcher(brokerController,
liteSubscriptionRegistry, liteLifecycleManager);
+ MessageArrivingListener listener = new NotifyMessageArrivingListener(
+ Mockito.mock(PullRequestHoldService.class),
Mockito.mock(PopMessageProcessor.class),
+ Mockito.mock(NotificationProcessor.class), liteEventDispatcher);
+
+ messageStore = LiteTestUtil.buildMessageStore(storePathRootDir,
BROKER_CONFIG, TOPIC_CONFIG_TABLE, false, listener);
+ when(brokerController.getMessageStore()).thenReturn(messageStore);
+ messageStore.load();
+ messageStore.start();
+
liteLifecycleManager.init();
}
@@ -112,6 +134,67 @@ public class LiteLifecycleManagerTest {
Assert.assertEquals(0,
liteLifecycleManager.getMaxOffsetInQueue(UUID.randomUUID().toString()));
}
+ @Test
+ public void testCollectByParentTopic() {
+ int num = 3;
+ String parentTopic = UUID.randomUUID().toString();
+ for (int i = 0; i < num; i++) {
+ messageStore.putMessage(LiteTestUtil.buildMessage(parentTopic,
UUID.randomUUID().toString()));
+
messageStore.putMessage(LiteTestUtil.buildMessage(UUID.randomUUID().toString(),
UUID.randomUUID().toString()));
+ }
+ await().atMost(5, SECONDS).pollInterval(200, MILLISECONDS).until(() ->
messageStore.dispatchBehindBytes() <= 0);
+ List<String> result =
liteLifecycleManager.collectByParentTopic(parentTopic);
+ Assert.assertEquals(num, result.size());
+ for (String lmqName : result) {
+ Assert.assertTrue(LiteUtil.belongsTo(lmqName, parentTopic));
+ }
+
+ result =
liteLifecycleManager.collectByParentTopic(UUID.randomUUID().toString());
+ Assert.assertEquals(0, result.size());
+ }
+
+ @Test
+ public void testGetLiteTopicCount() {
+ int num = 3;
+ String parentTopic = UUID.randomUUID().toString();
+ mockTopicConfig.getAttributes().put(
+ TopicAttributes.TOPIC_MESSAGE_TYPE_ATTRIBUTE.getName(),
TopicMessageType.LITE.getValue());
+ for (int i = 0; i < num; i++) {
+ messageStore.putMessage(LiteTestUtil.buildMessage(parentTopic,
UUID.randomUUID().toString()));
+
messageStore.putMessage(LiteTestUtil.buildMessage(UUID.randomUUID().toString(),
UUID.randomUUID().toString()));
+ }
+ await().atMost(5, SECONDS).pollInterval(200, MILLISECONDS).until(() ->
messageStore.dispatchBehindBytes() <= 0);
+
+ Assert.assertEquals(num,
liteLifecycleManager.getLiteTopicCount(parentTopic));
+ Assert.assertEquals(0,
liteLifecycleManager.getLiteTopicCount(UUID.randomUUID().toString()));
+ }
+
+ @Test
+ public void testCleanByParentTopic() {
+ int num = 3;
+ String parentTopic = UUID.randomUUID().toString();
+ mockTopicConfig.getAttributes().put(
+ TopicAttributes.TOPIC_MESSAGE_TYPE_ATTRIBUTE.getName(),
TopicMessageType.LITE.getValue());
+ List<String> liteTopics =
+ IntStream.range(0, num).mapToObj(i ->
UUID.randomUUID().toString()).collect(Collectors.toList());
+ for (int i = 0; i < num; i++) {
+ messageStore.putMessage(LiteTestUtil.buildMessage(parentTopic,
liteTopics.get(i)));
+ }
+ await().atMost(5, SECONDS).pollInterval(200, MILLISECONDS).until(() ->
messageStore.dispatchBehindBytes() <= 0);
+
+ for (int i = 0; i < num; i++) {
+ String lmqName = LiteUtil.toLmqName(parentTopic,
liteTopics.get(i));
+
Assert.assertTrue(messageStore.getQueueStore().getConsumeQueueTable().containsKey(lmqName));
+ }
+
+ liteLifecycleManager.cleanByParentTopic(parentTopic);
+
+ for (int i = 0; i < num; i++) {
+ String lmqName = LiteUtil.toLmqName(parentTopic,
liteTopics.get(i));
+
Assert.assertFalse(messageStore.getQueueStore().getConsumeQueueTable().containsKey(lmqName));
+ }
+ }
+
@Ignore
@Test
public void testCleanExpiredLiteTopic() {
diff --git
a/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteTestUtil.java
b/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteTestUtil.java
index ec6efb1fd5..2a1a3ce1f8 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteTestUtil.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteTestUtil.java
@@ -26,6 +26,7 @@ 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.store.DefaultMessageStore;
+import org.apache.rocketmq.store.MessageArrivingListener;
import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.RocksDBMessageStore;
import org.apache.rocketmq.store.config.FlushDiskType;
@@ -40,19 +41,31 @@ public class LiteTestUtil {
public static MessageStore buildMessageStore(final BrokerConfig
brokerConfig,
MessageStoreConfig storeConfig, final ConcurrentMap<String,
TopicConfig> topicConfigTable,
boolean isRocksDBStore) throws Exception {
+ return buildMessageStore(brokerConfig, storeConfig, topicConfigTable,
isRocksDBStore, null);
+ }
+
+ public static MessageStore buildMessageStore(final BrokerConfig
brokerConfig,
+ MessageStoreConfig storeConfig, final ConcurrentMap<String,
TopicConfig> topicConfigTable,
+ boolean isRocksDBStore, MessageArrivingListener
messageArrivingListener) throws Exception {
BrokerStatsManager brokerStatsManager = new
BrokerStatsManager(brokerConfig);
MessageStore messageStore;
if (isRocksDBStore) {
- messageStore = new RocksDBMessageStore(storeConfig,
brokerStatsManager, null, brokerConfig, topicConfigTable);
+ messageStore = new RocksDBMessageStore(storeConfig,
brokerStatsManager, messageArrivingListener, brokerConfig, topicConfigTable);
} else {
- messageStore = new DefaultMessageStore(storeConfig,
brokerStatsManager, null, brokerConfig, topicConfigTable);
+ messageStore = new DefaultMessageStore(storeConfig,
brokerStatsManager, messageArrivingListener, brokerConfig, topicConfigTable);
}
return messageStore;
}
public static MessageStore buildMessageStore(String storePathRootDir,
final BrokerConfig brokerConfig,
final ConcurrentMap<String, TopicConfig> topicConfigTable, boolean
isRocksDBStore) throws Exception {
+ return buildMessageStore(storePathRootDir, brokerConfig,
topicConfigTable, isRocksDBStore, null);
+ }
+
+ public static MessageStore buildMessageStore(String storePathRootDir,
final BrokerConfig brokerConfig,
+ final ConcurrentMap<String, TopicConfig> topicConfigTable, boolean
isRocksDBStore,
+ MessageArrivingListener messageArrivingListener) throws Exception {
MessageStoreConfig storeConfig = new MessageStoreConfig();
storeConfig.setMappedFileSizeCommitLog(1024 * 1024 * 10);
storeConfig.setMappedFileSizeConsumeQueue(1024 * 1024 * 10);
@@ -65,7 +78,7 @@ public class LiteTestUtil {
storeConfig.setEnableMultiDispatch(true);
storeConfig.setStorePathRootDir(storePathRootDir);
- return buildMessageStore(brokerConfig, storeConfig, topicConfigTable,
isRocksDBStore);
+ return buildMessageStore(brokerConfig, storeConfig, topicConfigTable,
isRocksDBStore, messageArrivingListener);
}
public static MessageExtBrokerInner buildMessage(String parentTopic,
String liteTopic) {
diff --git
a/broker/src/test/java/org/apache/rocketmq/broker/lite/RocksDBLiteLifecycleManagerTest.java
b/broker/src/test/java/org/apache/rocketmq/broker/lite/RocksDBLiteLifecycleManagerTest.java
index dd34840535..bbc1f4d431 100644
---
a/broker/src/test/java/org/apache/rocketmq/broker/lite/RocksDBLiteLifecycleManagerTest.java
+++
b/broker/src/test/java/org/apache/rocketmq/broker/lite/RocksDBLiteLifecycleManagerTest.java
@@ -18,12 +18,20 @@
package org.apache.rocketmq.broker.lite;
import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.broker.longpolling.NotifyMessageArrivingListener;
+import org.apache.rocketmq.broker.longpolling.PullRequestHoldService;
+import org.apache.rocketmq.broker.offset.ConsumerOffsetManager;
+import org.apache.rocketmq.broker.processor.NotificationProcessor;
+import org.apache.rocketmq.broker.processor.PopMessageProcessor;
import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
import org.apache.rocketmq.broker.topic.TopicConfigManager;
import org.apache.rocketmq.common.BrokerConfig;
+import org.apache.rocketmq.common.TopicAttributes;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.UtilAll;
+import org.apache.rocketmq.common.attribute.TopicMessageType;
import org.apache.rocketmq.common.lite.LiteUtil;
+import org.apache.rocketmq.store.MessageArrivingListener;
import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.plugin.AbstractPluginMessageStore;
@@ -40,6 +48,7 @@ import org.mockito.Mockito;
import org.mockito.junit.MockitoJUnitRunner;
import java.io.File;
+import java.util.Collections;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
@@ -53,9 +62,10 @@ import static org.awaitility.Awaitility.await;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.nullable;
import static org.mockito.Mockito.when;
-@RunWith(MockitoJUnitRunner.class)
+@RunWith(MockitoJUnitRunner.Silent.class)
public class RocksDBLiteLifecycleManagerTest {
private final static BrokerConfig BROKER_CONFIG = new BrokerConfig();
@@ -63,30 +73,51 @@ public class RocksDBLiteLifecycleManagerTest {
private static String storePathRootDir;
private static MessageStore messageStore;
private static RocksDBLiteLifecycleManager liteLifecycleManager;
+ private static LiteEventDispatcher liteEventDispatcher;
+ private static TopicConfig mockTopicConfig = new TopicConfig();
@BeforeClass
public static void setUp() throws Exception {
storePathRootDir = System.getProperty("java.io.tmpdir") +
File.separator + "store-rocksDBLifecycleTest";
UtilAll.deleteFile(new File(storePathRootDir));
- messageStore = LiteTestUtil.buildMessageStore(storePathRootDir,
BROKER_CONFIG, TOPIC_CONFIG_TABLE, true);
- messageStore.load();
- messageStore.start();
-
BrokerController brokerController =
Mockito.mock(BrokerController.class);
LiteSharding liteSharding = Mockito.mock(LiteSharding.class);
TopicConfigManager topicConfigManager =
Mockito.mock(TopicConfigManager.class);
SubscriptionGroupManager subscriptionGroupManager =
Mockito.mock(SubscriptionGroupManager.class);
+ LiteSubscriptionRegistry liteSubscriptionRegistry =
Mockito.mock(LiteSubscriptionRegistry.class);
+ ConsumerOffsetManager consumerOffsetManager =
Mockito.mock(ConsumerOffsetManager.class);
+ when(consumerOffsetManager.getOffsetTable()).thenReturn(new
ConcurrentHashMap<>());
+ when(consumerOffsetManager.getPullOffsetTable()).thenReturn(new
ConcurrentHashMap<>());
+ // dispatch() fans out to subscribers right after maintaining the
prefix index; no subscriber here
+
when(liteSubscriptionRegistry.getAllSubscribers(nullable(String.class),
anyString()))
+ .thenReturn(Collections.emptyMap());
when(brokerController.getBrokerConfig()).thenReturn(BROKER_CONFIG);
- when(brokerController.getMessageStore()).thenReturn(messageStore);
when(brokerController.getTopicConfigManager()).thenReturn(topicConfigManager);
when(brokerController.getSubscriptionGroupManager()).thenReturn(subscriptionGroupManager);
+
when(brokerController.getLiteSubscriptionRegistry()).thenReturn(liteSubscriptionRegistry);
+
when(brokerController.getConsumerOffsetManager()).thenReturn(consumerOffsetManager);
when(topicConfigManager.getTopicConfigTable()).thenReturn(TOPIC_CONFIG_TABLE);
+
when(topicConfigManager.selectTopicConfig(anyString())).thenReturn(mockTopicConfig);
when(subscriptionGroupManager.getSubscriptionGroupTable()).thenReturn(new
ConcurrentHashMap<>());
RocksDBLiteLifecycleManager testObject = new
RocksDBLiteLifecycleManager(brokerController, liteSharding);
liteLifecycleManager = Mockito.spy(testObject);
+
+ // Wire the real notify path so putMessage drives the prefix index.
RocksDB CQ is committed by
+ // RocksGroupCommitService (isNotifyMessageArriveWhenReput()==false),
but it still lands on the same
+ // MessageArrivingListener -> LiteEventDispatcher.dispatch ->
onLmqCreate path.
+ liteEventDispatcher = new LiteEventDispatcher(brokerController,
liteSubscriptionRegistry, liteLifecycleManager);
+ MessageArrivingListener listener = new NotifyMessageArrivingListener(
+ Mockito.mock(PullRequestHoldService.class),
Mockito.mock(PopMessageProcessor.class),
+ Mockito.mock(NotificationProcessor.class), liteEventDispatcher);
+
+ messageStore = LiteTestUtil.buildMessageStore(storePathRootDir,
BROKER_CONFIG, TOPIC_CONFIG_TABLE, true, listener);
+ when(brokerController.getMessageStore()).thenReturn(messageStore);
+ messageStore.load();
+ messageStore.start();
+
liteLifecycleManager.init();
}
@@ -95,6 +126,7 @@ public class RocksDBLiteLifecycleManagerTest {
messageStore.shutdown();
messageStore.destroy();
UtilAll.deleteFile(new File(storePathRootDir));
+ mockTopicConfig = new TopicConfig();
}
@Ignore
@@ -141,6 +173,69 @@ public class RocksDBLiteLifecycleManagerTest {
Assert.assertEquals(0,
liteLifecycleManager.getMaxOffsetInQueue(UUID.randomUUID().toString()));
}
+ @Test
+ public void testCollectByParentTopic() {
+ int num = 3;
+ String parentTopic = UUID.randomUUID().toString();
+ for (int i = 0; i < num; i++) {
+ messageStore.putMessage(LiteTestUtil.buildMessage(parentTopic,
UUID.randomUUID().toString()));
+
messageStore.putMessage(LiteTestUtil.buildMessage(UUID.randomUUID().toString(),
UUID.randomUUID().toString()));
+ }
+ await().atMost(5, SECONDS).pollInterval(200, MILLISECONDS).until(() ->
messageStore.dispatchBehindBytes() <= 0);
+ List<String> result =
liteLifecycleManager.collectByParentTopic(parentTopic);
+ Assert.assertEquals(num, result.size());
+ for (String lmqName : result) {
+ Assert.assertTrue(LiteUtil.belongsTo(lmqName, parentTopic));
+ }
+
+ result =
liteLifecycleManager.collectByParentTopic(UUID.randomUUID().toString());
+ Assert.assertEquals(0, result.size());
+ }
+
+ @Test
+ public void testGetLiteTopicCount() {
+ int num = 3;
+ String parentTopic = UUID.randomUUID().toString();
+ mockTopicConfig.getAttributes().put(
+ TopicAttributes.TOPIC_MESSAGE_TYPE_ATTRIBUTE.getName(),
TopicMessageType.LITE.getValue());
+ for (int i = 0; i < num; i++) {
+ messageStore.putMessage(LiteTestUtil.buildMessage(parentTopic,
UUID.randomUUID().toString()));
+
messageStore.putMessage(LiteTestUtil.buildMessage(UUID.randomUUID().toString(),
UUID.randomUUID().toString()));
+ }
+ await().atMost(5, SECONDS).pollInterval(200, MILLISECONDS).until(() ->
messageStore.dispatchBehindBytes() <= 0);
+
+ Assert.assertEquals(num,
liteLifecycleManager.getLiteTopicCount(parentTopic));
+ Assert.assertEquals(0,
liteLifecycleManager.getLiteTopicCount(UUID.randomUUID().toString()));
+ }
+
+ @Test
+ public void testCleanByParentTopic() throws Exception {
+ int num = 3;
+ String parentTopic = UUID.randomUUID().toString();
+ mockTopicConfig.getAttributes().put(
+ TopicAttributes.TOPIC_MESSAGE_TYPE_ATTRIBUTE.getName(),
TopicMessageType.LITE.getValue());
+ List<String> liteTopics =
+ IntStream.range(0, num).mapToObj(i ->
UUID.randomUUID().toString()).collect(Collectors.toList());
+ for (int i = 0; i < num; i++) {
+ messageStore.putMessage(LiteTestUtil.buildMessage(parentTopic,
liteTopics.get(i)));
+ }
+ await().atMost(5, SECONDS).pollInterval(200, MILLISECONDS).until(() ->
messageStore.dispatchBehindBytes() <= 0);
+
+ for (int i = 0; i < num; i++) {
+ String lmqName = LiteUtil.toLmqName(parentTopic,
liteTopics.get(i));
+ Assert.assertEquals(1, (long)
messageStore.getQueueStore().getMaxOffset(lmqName, 0));
+ Assert.assertEquals(1,
liteLifecycleManager.getMaxOffsetInQueue(lmqName));
+ }
+
+ liteLifecycleManager.cleanByParentTopic(parentTopic);
+
+ for (int i = 0; i < num; i++) {
+ String lmqName = LiteUtil.toLmqName(parentTopic,
liteTopics.get(i));
+ Assert.assertEquals(0, (long)
messageStore.getQueueStore().getMaxOffset(lmqName, 0));
+ Assert.assertEquals(0,
liteLifecycleManager.getMaxOffsetInQueue(lmqName));
+ }
+ }
+
@Test
public void testCleanExpiredLiteTopic() throws Exception {
int num = 3;
