support storing Retained messages of single tenant cross multiple ranges

bifromqio · Sep 20, 2024 · dcb8f6e · dcb8f6e
1 parent 0a92a32
commit dcb8f6e
Show file tree

Hide file tree

Showing 57 changed files with 1,215 additions and 708 deletions.
diff --git a/...t/bifromq-dist-rpc-definition/src/main/java/com/baidu/bifromq/dist/entity/EntityUtil.java b/...t/bifromq-dist-rpc-definition/src/main/java/com/baidu/bifromq/dist/entity/EntityUtil.java
@@ -13,14 +13,14 @@
 
 package com.baidu.bifromq.dist.entity;
 
-import static com.baidu.bifromq.dist.util.TopicUtil.escape;
-import static com.baidu.bifromq.dist.util.TopicUtil.isNormalTopicFilter;
-import static com.baidu.bifromq.dist.util.TopicUtil.unescape;
+import static com.baidu.bifromq.util.TopicUtil.escape;
+import static com.baidu.bifromq.util.TopicUtil.unescape;
 import static com.baidu.bifromq.util.TopicConst.DELIMITER_CHAR;
 import static com.baidu.bifromq.util.TopicConst.NUL;
 import static com.baidu.bifromq.util.TopicConst.NUL_CHAR;
 import static com.baidu.bifromq.util.TopicConst.ORDERED_SHARE;
 import static com.baidu.bifromq.util.TopicConst.UNORDERED_SHARE;
+import static com.baidu.bifromq.util.TopicUtil.isNormalTopicFilter;
 import static com.google.protobuf.ByteString.copyFromUtf8;
 
 import com.baidu.bifromq.dist.rpc.proto.GroupMatchRecord;

diff --git a/...ifromq-dist-rpc-definition/src/main/java/com/baidu/bifromq/dist/entity/GroupMatching.java b/...ifromq-dist-rpc-definition/src/main/java/com/baidu/bifromq/dist/entity/GroupMatching.java
@@ -16,7 +16,7 @@
 import static com.baidu.bifromq.util.TopicConst.DELIMITER_CHAR;
 import static com.baidu.bifromq.util.TopicConst.ORDERED_SHARE;
 import static com.baidu.bifromq.util.TopicConst.UNORDERED_SHARE;
-import static com.baidu.bifromq.dist.util.TopicUtil.unescape;
+import static com.baidu.bifromq.util.TopicUtil.unescape;
 
 import com.google.common.collect.Sets;
 import com.google.protobuf.ByteString;

diff --git a/...fromq-dist-rpc-definition/src/main/java/com/baidu/bifromq/dist/entity/NormalMatching.java b/...fromq-dist-rpc-definition/src/main/java/com/baidu/bifromq/dist/entity/NormalMatching.java
@@ -14,7 +14,7 @@
 package com.baidu.bifromq.dist.entity;
 
 
-import static com.baidu.bifromq.dist.util.TopicUtil.unescape;
+import static com.baidu.bifromq.util.TopicUtil.unescape;
 import static com.baidu.bifromq.util.TopicConst.NUL;
 
 import com.baidu.bifromq.type.MatchInfo;

diff --git a/...dist/bifromq-dist-rpc-definition/src/main/java/com/baidu/bifromq/dist/util/TopicUtil.java b/...dist/bifromq-dist-rpc-definition/src/main/java/com/baidu/bifromq/dist/util/TopicUtil.java
diff --git a/bifromq-dist/bifromq-dist-rpc-definition/src/test/java/com/baidu/bifromq/dist/TestUtil.java b/bifromq-dist/bifromq-dist-rpc-definition/src/test/java/com/baidu/bifromq/dist/TestUtil.java
@@ -13,8 +13,8 @@
 
 package com.baidu.bifromq.dist;
 
-import static com.baidu.bifromq.dist.util.TopicUtil.fastJoin;
-import static com.baidu.bifromq.dist.util.TopicUtil.parse;
+import static com.baidu.bifromq.util.TopicUtil.fastJoin;
+import static com.baidu.bifromq.util.TopicUtil.parse;
 import static com.baidu.bifromq.util.TopicConst.DELIMITER;
 import static com.baidu.bifromq.util.TopicConst.MULTI_WILDCARD;
 import static com.baidu.bifromq.util.TopicConst.NUL;

diff --git a/...q-dist/bifromq-dist-rpc-definition/src/test/java/com/baidu/bifromq/dist/TopicMatcher.java b/...q-dist/bifromq-dist-rpc-definition/src/test/java/com/baidu/bifromq/dist/TopicMatcher.java
@@ -19,7 +19,7 @@
 import static com.baidu.bifromq.util.TopicConst.SINGLE_WILDCARD;
 import static com.baidu.bifromq.util.TopicConst.SYS_PREFIX;
 
-import com.baidu.bifromq.dist.util.TopicUtil;
+import com.baidu.bifromq.util.TopicUtil;
 import java.util.List;
 import java.util.Optional;
 

diff --git a/...st/bifromq-dist-rpc-definition/src/test/java/com/baidu/bifromq/dist/TopicMatcherTest.java b/...st/bifromq-dist-rpc-definition/src/test/java/com/baidu/bifromq/dist/TopicMatcherTest.java
@@ -13,7 +13,7 @@
 
 package com.baidu.bifromq.dist;
 
-import static com.baidu.bifromq.dist.util.TopicUtil.escape;
+import static com.baidu.bifromq.util.TopicUtil.escape;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertTrue;

diff --git a/...fromq-dist-rpc-definition/src/test/java/com/baidu/bifromq/dist/entity/EntityUtilTest.java b/...fromq-dist-rpc-definition/src/test/java/com/baidu/bifromq/dist/entity/EntityUtilTest.java
@@ -17,7 +17,7 @@
 import static com.baidu.bifromq.dist.entity.EntityUtil.parseTopicFilter;
 import static com.baidu.bifromq.dist.entity.EntityUtil.toMatchRecordKey;
 import static com.baidu.bifromq.dist.entity.EntityUtil.toQInboxId;
-import static com.baidu.bifromq.dist.util.TopicUtil.escape;
+import static com.baidu.bifromq.util.TopicUtil.escape;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertTrue;
 

diff --git a/...ist-rpc-definition/src/test/java/com/baidu/bifromq/dist/trie/TopicFilterIteratorTest.java b/...ist-rpc-definition/src/test/java/com/baidu/bifromq/dist/trie/TopicFilterIteratorTest.java
@@ -15,15 +15,15 @@
 
 import static com.baidu.bifromq.dist.TestUtil.randomTopic;
 import static com.baidu.bifromq.dist.TestUtil.randomTopicFilter;
-import static com.baidu.bifromq.dist.util.TopicUtil.escape;
-import static com.baidu.bifromq.dist.util.TopicUtil.fastJoin;
-import static com.baidu.bifromq.dist.util.TopicUtil.parse;
+import static com.baidu.bifromq.util.TopicUtil.escape;
+import static com.baidu.bifromq.util.TopicUtil.fastJoin;
+import static com.baidu.bifromq.util.TopicUtil.parse;
 import static com.baidu.bifromq.util.TopicConst.NUL;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertTrue;
 
 import com.baidu.bifromq.dist.TestUtil;
-import com.baidu.bifromq.dist.util.TopicUtil;
+import com.baidu.bifromq.util.TopicUtil;
 import com.baidu.bifromq.util.TopicConst;
 import java.util.ArrayList;
 import java.util.HashSet;

diff --git a/...n/src/test/java/com/baidu/bifromq/dist/trie/benchmark/TopicTrieBuilderBenchmarkState.java b/...n/src/test/java/com/baidu/bifromq/dist/trie/benchmark/TopicTrieBuilderBenchmarkState.java
@@ -15,7 +15,7 @@
 
 import com.baidu.bifromq.dist.trie.TopicTrieNode;
 import com.baidu.bifromq.dist.TestUtil;
-import com.baidu.bifromq.dist.util.TopicUtil;
+import com.baidu.bifromq.util.TopicUtil;
 import java.util.List;
 import lombok.extern.slf4j.Slf4j;
 import org.openjdk.jmh.annotations.Level;

diff --git a/.../bifromq-dist-rpc-definition/src/test/java/com/baidu/bifromq/dist/util/TopicUtilTest.java b/.../bifromq-dist-rpc-definition/src/test/java/com/baidu/bifromq/dist/util/TopicUtilTest.java
diff --git a/...-dist-server/src/main/java/com/baidu/bifromq/dist/server/scheduler/DistCallScheduler.java b/...-dist-server/src/main/java/com/baidu/bifromq/dist/server/scheduler/DistCallScheduler.java
@@ -39,7 +39,7 @@
 import com.baidu.bifromq.dist.rpc.proto.DistServiceROCoProcInput;
 import com.baidu.bifromq.dist.trie.TopicFilterIterator;
 import com.baidu.bifromq.dist.trie.TopicTrieNode;
-import com.baidu.bifromq.dist.util.TopicUtil;
+import com.baidu.bifromq.util.TopicUtil;
 import com.baidu.bifromq.plugin.settingprovider.ISettingProvider;
 import com.baidu.bifromq.plugin.settingprovider.Setting;
 import com.baidu.bifromq.sysprops.props.DataPlaneBurstLatencyMillis;

diff --git a/...bifromq-dist-worker/src/main/java/com/baidu/bifromq/dist/worker/DeliverExecutorGroup.java b/...bifromq-dist-worker/src/main/java/com/baidu/bifromq/dist/worker/DeliverExecutorGroup.java
@@ -13,7 +13,7 @@
 
 package com.baidu.bifromq.dist.worker;
 
-import static com.baidu.bifromq.dist.util.TopicUtil.escape;
+import static com.baidu.bifromq.util.TopicUtil.escape;
 import static com.baidu.bifromq.metrics.TenantMetric.MqttPersistentFanOutBytes;
 import static com.baidu.bifromq.plugin.eventcollector.ThreadLocalEventPool.getLocal;
 import static com.bifromq.plugin.resourcethrottler.TenantResourceType.TotalPersistentFanOutBytesPerSeconds;

diff --git a/...ist/bifromq-dist-worker/src/main/java/com/baidu/bifromq/dist/worker/DistWorkerCoProc.java b/...ist/bifromq-dist-worker/src/main/java/com/baidu/bifromq/dist/worker/DistWorkerCoProc.java
@@ -28,7 +28,7 @@
 import static com.baidu.bifromq.dist.entity.EntityUtil.toGroupMatchRecordKey;
 import static com.baidu.bifromq.dist.entity.EntityUtil.toNormalMatchRecordKey;
 import static com.baidu.bifromq.dist.entity.EntityUtil.toScopedTopicFilter;
-import static com.baidu.bifromq.dist.util.TopicUtil.isNormalTopicFilter;
+import static com.baidu.bifromq.util.TopicUtil.isNormalTopicFilter;
 import static java.util.Collections.singletonMap;
 
 import com.baidu.bifromq.basekv.proto.Boundary;

diff --git a/...bifromq/dist/worker/index/TopicIndex.java → ...baidu/bifromq/dist/worker/TopicIndex.java b/...bifromq/dist/worker/index/TopicIndex.java → ...baidu/bifromq/dist/worker/TopicIndex.java
@@ -11,13 +11,15 @@
  * See the License for the specific language governing permissions and limitations under the License.
  */
 
-package com.baidu.bifromq.dist.worker.index;
+package com.baidu.bifromq.dist.worker;
 
 import static com.baidu.bifromq.util.TopicConst.MULTI_WILDCARD;
 import static com.baidu.bifromq.util.TopicConst.SINGLE_WILDCARD;
 import static com.baidu.bifromq.util.TopicConst.SYS_PREFIX;
 
-import com.baidu.bifromq.dist.util.TopicUtil;
+import com.baidu.bifromq.util.TopicUtil;
+import com.baidu.bifromq.util.index.Branch;
+import com.baidu.bifromq.util.index.TopicLevelTrie;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -35,6 +37,8 @@ public <T> Map<Branch<T>, Action> selectBranch(Map<String, Branch<T>> branches,
             if (currentLevel < topicLevels.size() - 1) {
                 // not last level
                 String topicLevelToMatch = topicLevels.get(currentLevel);
+                boolean matchParent = currentLevel + 1 == topicLevels.size() - 1
+                    && topicLevels.get(currentLevel + 1).equals(MULTI_WILDCARD);
                 switch (topicLevelToMatch) {
                     case SINGLE_WILDCARD -> {
                         Map<Branch<T>, Action> result = new HashMap<>();
@@ -44,30 +48,15 @@ public <T> Map<Branch<T>, Action> selectBranch(Map<String, Branch<T>> branches,
                                 // + skip SYS topic
                                 continue;
                             }
-                            result.put(branch, Action.CONTINUE);
-                        }
-                        return result;
-                    }
-                    case MULTI_WILDCARD -> {
-                        Map<Branch<T>, Action> result = new HashMap<>();
-                        for (Map.Entry<String, Branch<T>> entry : branches.entrySet()) {
-                            Branch<T> branch = entry.getValue();
-                            if (currentLevel == 0 && entry.getKey().startsWith(SYS_PREFIX)) {
-                                // # skip SYS topic
-                                continue;
-                            }
-                            result.put(branch, Action.MATCH_AND_CONTINUE);
+                            result.put(branch, matchParent ? Action.MATCH_AND_CONTINUE : Action.CONTINUE);
                         }
                         return result;
                     }
                     default -> {
+                        assert !topicLevelToMatch.equals(MULTI_WILDCARD) : "MULTI_WILDCARD should be the last level";
                         if (branches.containsKey(topicLevelToMatch)) {
-                            if (currentLevel + 1 < topicLevels.size()
-                                && topicLevels.get(currentLevel + 1).equals(MULTI_WILDCARD)) {
-                                // # match parent level
-                                return Map.of(branches.get(topicLevelToMatch), Action.MATCH_AND_CONTINUE);
-                            }
-                            return Map.of(branches.get(topicLevelToMatch), Action.CONTINUE);
+                            return Map.of(branches.get(topicLevelToMatch),
+                                matchParent ? Action.MATCH_AND_CONTINUE : Action.CONTINUE);
                         }
                         return Collections.emptyMap();
                     }

diff --git a/...fromq-dist-worker/src/main/java/com/baidu/bifromq/dist/worker/cache/TenantRouteCache.java b/...fromq-dist-worker/src/main/java/com/baidu/bifromq/dist/worker/cache/TenantRouteCache.java
@@ -16,10 +16,10 @@
 import com.baidu.bifromq.basekv.proto.Boundary;
 import com.baidu.bifromq.dist.entity.GroupMatching;
 import com.baidu.bifromq.dist.entity.Matching;
-import com.baidu.bifromq.dist.worker.index.TopicIndex;
 import com.baidu.bifromq.metrics.ITenantMeter;
 import com.baidu.bifromq.metrics.TenantMetric;
 import com.baidu.bifromq.sysprops.props.DistMaxCachedRoutesPerTenant;
+import com.baidu.bifromq.dist.worker.TopicIndex;
 import com.github.benmanes.caffeine.cache.Cache;
 import com.github.benmanes.caffeine.cache.Caffeine;
 import com.github.benmanes.caffeine.cache.Expiry;

diff --git a/...omq-dist-worker/src/main/java/com/baidu/bifromq/dist/worker/cache/TenantRouteMatcher.java b/...omq-dist-worker/src/main/java/com/baidu/bifromq/dist/worker/cache/TenantRouteMatcher.java
@@ -24,7 +24,7 @@
 import com.baidu.bifromq.dist.entity.Matching;
 import com.baidu.bifromq.dist.trie.TopicFilterIterator;
 import com.baidu.bifromq.dist.trie.TopicTrieNode;
-import com.baidu.bifromq.dist.util.TopicUtil;
+import com.baidu.bifromq.util.TopicUtil;
 import com.google.common.collect.Sets;
 import com.google.protobuf.ByteString;
 import io.micrometer.core.instrument.Timer;