You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@bifromq.apache.org by po...@apache.org on 2025/08/14 06:36:08 UTC
(bifromq) branch main updated: [Feat] new runtime settings for limiting fanout scale to persistent sessions (#166)
This is an automated email from the ASF dual-hosted git repository.
popduke pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/bifromq.git
The following commit(s) were added to refs/heads/main by this push:
new 93eb4354 [Feat] new runtime settings for limiting fanout scale to persistent sessions (#166)
93eb4354 is described below
commit 93eb4354ae2c53436c735d48421f4bbf019999ff
Author: Yonny(Yu) Hao <po...@gmail.com>
AuthorDate: Thu Aug 14 14:36:03 2025 +0800
[Feat] new runtime settings for limiting fanout scale to persistent sessions (#166)
* Tenant-level settings for limiting group fanout and persistent fanout
* Report relevant events when throttling happens
* Reduced memory consumption in data plane
---
.../bifromq/deliverer/BatchDeliveryCall.java | 36 +--
.../dist/client/scheduler/BatchPubCall.java | 32 ++-
.../dist/server/scheduler/BatchDistServerCall.java | 4 +-
.../bifromq/dist/server/DistServiceTest.java | 11 +-
.../bifromq/dist/worker/schema/GroupMatching.java | 5 +-
.../bifromq/dist/worker/schema/Matching.java | 4 +-
.../bifromq/dist/worker/schema/NormalMatching.java | 6 +-
bifromq-dist/bifromq-dist-worker/pom.xml | 4 +
.../bifromq/dist/worker/DeliverExecutorGroup.java | 264 ++++++++++++--------
.../org/apache/bifromq/dist/worker/DistWorker.java | 5 +-
.../bifromq/dist/worker/DistWorkerBuilder.java | 26 +-
.../dist/worker/DistWorkerCoProcFactory.java | 48 ++--
...TenantRouteMatcher.java => IMatchedRoutes.java} | 45 +++-
.../dist/worker/cache/ITenantRouteMatcher.java | 7 +-
.../dist/worker/cache/SubscriptionCache.java | 33 ++-
.../dist/worker/cache/TenantRouteCache.java | 113 +++++++--
.../dist/worker/cache/TenantRouteCacheFactory.java | 18 +-
.../dist/worker/cache/TenantRouteMatcher.java | 147 +++++++++--
.../apache/bifromq/dist/worker/DistQoS0Test.java | 18 +-
.../apache/bifromq/dist/worker/DistWorkerTest.java | 58 +++--
.../bifromq/dist/worker/FanoutThrottledTest.java | 263 ++++++++++++++++++++
.../dist/worker/cache/TenantRouteCacheTest.java | 270 +++++++++++++++++----
.../apache/bifromq/mqtt/integration/MQTTTest.java | 36 +--
.../bifromq/plugin/eventcollector/EventType.java | 10 +-
.../eventcollector/distservice/DeliverNoInbox.java | 55 -----
...Unsubscribed.java => GroupFanoutThrottled.java} | 24 +-
...or.java => PersistentFanoutBytesThrottled.java} | 27 +--
...scribed.java => PersistentFanoutThrottled.java} | 29 +--
.../eventcollector/distservice/SubscribeError.java | 60 -----
.../bifromq/plugin/settingprovider/Setting.java | 5 +-
.../retain/server/scheduler/BatchMatchCall.java | 32 +--
...DistCachedRoutesFanoutCheckIntervalSeconds.java | 25 +-
.../bifromq/starter/module/DistServiceModule.java | 3 +-
pom.xml | 2 +-
34 files changed, 1191 insertions(+), 534 deletions(-)
diff --git a/bifromq-deliverer/src/main/java/org/apache/bifromq/deliverer/BatchDeliveryCall.java b/bifromq-deliverer/src/main/java/org/apache/bifromq/deliverer/BatchDeliveryCall.java
index 653db671..f39afe2c 100644
--- a/bifromq-deliverer/src/main/java/org/apache/bifromq/deliverer/BatchDeliveryCall.java
+++ b/bifromq-deliverer/src/main/java/org/apache/bifromq/deliverer/BatchDeliveryCall.java
@@ -14,7 +14,7 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
package org.apache.bifromq.deliverer;
@@ -26,26 +26,27 @@ import static org.apache.bifromq.deliverer.DeliveryCallResult.NO_SUB;
import static org.apache.bifromq.deliverer.DeliveryCallResult.OK;
import static org.apache.bifromq.plugin.subbroker.TypeUtil.toMap;
-import org.apache.bifromq.basescheduler.IBatchCall;
-import org.apache.bifromq.basescheduler.ICallTask;
-import org.apache.bifromq.dist.client.IDistClient;
-import org.apache.bifromq.plugin.subbroker.DeliveryPack;
-import org.apache.bifromq.plugin.subbroker.DeliveryPackage;
-import org.apache.bifromq.plugin.subbroker.DeliveryReply;
-import org.apache.bifromq.plugin.subbroker.DeliveryRequest;
-import org.apache.bifromq.plugin.subbroker.DeliveryResult;
-import org.apache.bifromq.plugin.subbroker.IDeliverer;
-import org.apache.bifromq.type.MatchInfo;
import java.util.ArrayDeque;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import lombok.extern.slf4j.Slf4j;
+import org.apache.bifromq.basescheduler.IBatchCall;
+import org.apache.bifromq.basescheduler.ICallTask;
+import org.apache.bifromq.dist.client.IDistClient;
+import org.apache.bifromq.plugin.subbroker.DeliveryPack;
+import org.apache.bifromq.plugin.subbroker.DeliveryPackage;
+import org.apache.bifromq.plugin.subbroker.DeliveryReply;
+import org.apache.bifromq.plugin.subbroker.DeliveryRequest;
+import org.apache.bifromq.plugin.subbroker.DeliveryResult;
+import org.apache.bifromq.plugin.subbroker.IDeliverer;
+import org.apache.bifromq.type.MatchInfo;
@Slf4j
class BatchDeliveryCall implements IBatchCall<DeliveryCall, DeliveryCallResult, DelivererKey> {
@@ -53,7 +54,7 @@ class BatchDeliveryCall implements IBatchCall<DeliveryCall, DeliveryCallResult,
private final IDeliverer deliverer;
private final DelivererKey batcherKey;
private final Queue<ICallTask<DeliveryCall, DeliveryCallResult, DelivererKey>> tasks = new ArrayDeque<>(128);
- private Map<String, Map<TopicMessagePackHolder, Set<MatchInfo>>> batch = new HashMap<>(128);
+ private final Map<String, Map<TopicMessagePackHolder, Set<MatchInfo>>> batch = new HashMap<>(128);
BatchDeliveryCall(IDistClient distClient, IDeliverer deliverer, DelivererKey batcherKey) {
this.distClient = distClient;
@@ -63,8 +64,6 @@ class BatchDeliveryCall implements IBatchCall<DeliveryCall, DeliveryCallResult,
@Override
public void reset() {
- batch = new HashMap<>(128);
- tasks.clear();
}
@Override
@@ -78,7 +77,11 @@ class BatchDeliveryCall implements IBatchCall<DeliveryCall, DeliveryCallResult,
@Override
public CompletableFuture<Void> execute() {
DeliveryRequest.Builder requestBuilder = DeliveryRequest.newBuilder();
- batch.forEach((tenantId, pack) -> {
+ Iterator<Map.Entry<String, Map<TopicMessagePackHolder, Set<MatchInfo>>>> itr = batch.entrySet().iterator();
+ while (itr.hasNext()) {
+ Map.Entry<String, Map<TopicMessagePackHolder, Set<MatchInfo>>> entry = itr.next();
+ String tenantId = entry.getKey();
+ Map<TopicMessagePackHolder, Set<MatchInfo>> pack = entry.getValue();
DeliveryPackage.Builder packageBuilder = DeliveryPackage.newBuilder();
pack.forEach((msgPackWrapper, matchInfos) -> {
DeliveryPack.Builder packBuilder = DeliveryPack.newBuilder().setMessagePack(msgPackWrapper.messagePack);
@@ -86,7 +89,8 @@ class BatchDeliveryCall implements IBatchCall<DeliveryCall, DeliveryCallResult,
packageBuilder.addPack(packBuilder.build());
});
requestBuilder.putPackage(tenantId, packageBuilder.build());
- });
+ itr.remove();
+ }
DeliveryRequest request = requestBuilder.build();
return execute(request);
}
diff --git a/bifromq-dist/bifromq-dist-client/src/main/java/org/apache/bifromq/dist/client/scheduler/BatchPubCall.java b/bifromq-dist/bifromq-dist-client/src/main/java/org/apache/bifromq/dist/client/scheduler/BatchPubCall.java
index 20d88774..39ac0b18 100644
--- a/bifromq-dist/bifromq-dist-client/src/main/java/org/apache/bifromq/dist/client/scheduler/BatchPubCall.java
+++ b/bifromq-dist/bifromq-dist-client/src/main/java/org/apache/bifromq/dist/client/scheduler/BatchPubCall.java
@@ -14,11 +14,17 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
package org.apache.bifromq.dist.client.scheduler;
+import java.util.ArrayDeque;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
import org.apache.bifromq.base.util.AsyncRetry;
import org.apache.bifromq.base.util.exception.NeedRetryException;
import org.apache.bifromq.baserpc.client.IRPCClient;
@@ -29,17 +35,12 @@ import org.apache.bifromq.dist.rpc.proto.DistReply;
import org.apache.bifromq.dist.rpc.proto.DistRequest;
import org.apache.bifromq.type.ClientInfo;
import org.apache.bifromq.type.PublisherMessagePack;
-import java.util.ArrayDeque;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Queue;
-import java.util.concurrent.CompletableFuture;
class BatchPubCall implements IBatchCall<PubRequest, PubResult, PubCallBatcherKey> {
private final IRPCClient.IRequestPipeline<DistRequest, DistReply> ppln;
private final Queue<ICallTask<PubRequest, PubResult, PubCallBatcherKey>> tasks = new ArrayDeque<>(64);
private final long retryTimeoutNanos;
- private Map<ClientInfo, Map<String, PublisherMessagePack.TopicPack.Builder>> clientMsgPack = new HashMap<>(128);
+ private final Map<ClientInfo, Map<String, PublisherMessagePack.TopicPack.Builder>> clientMsgPack = new HashMap<>(128);
BatchPubCall(IRPCClient.IRequestPipeline<DistRequest, DistReply> ppln, long retryTimeoutNanos) {
this.ppln = ppln;
@@ -48,8 +49,6 @@ class BatchPubCall implements IBatchCall<PubRequest, PubResult, PubCallBatcherKe
@Override
public void reset() {
- clientMsgPack = new HashMap<>(128);
- tasks.clear();
}
@Override
@@ -64,13 +63,20 @@ class BatchPubCall implements IBatchCall<PubRequest, PubResult, PubCallBatcherKe
@Override
public CompletableFuture<Void> execute() {
DistRequest.Builder requestBuilder = DistRequest.newBuilder().setReqId(System.nanoTime());
- clientMsgPack.forEach((k, v) -> {
- PublisherMessagePack.Builder senderMsgPackBuilder = PublisherMessagePack.newBuilder().setPublisher(k);
- for (PublisherMessagePack.TopicPack.Builder packBuilder : v.values()) {
+ Iterator<Map.Entry<ClientInfo, Map<String, PublisherMessagePack.TopicPack.Builder>>> itr =
+ clientMsgPack.entrySet().iterator();
+ while (itr.hasNext()) {
+ Map.Entry<ClientInfo, Map<String, PublisherMessagePack.TopicPack.Builder>> entry = itr.next();
+ ClientInfo publisher = entry.getKey();
+ Map<String, PublisherMessagePack.TopicPack.Builder> topicPackMap = entry.getValue();
+ PublisherMessagePack.Builder senderMsgPackBuilder = PublisherMessagePack.newBuilder()
+ .setPublisher(publisher);
+ for (PublisherMessagePack.TopicPack.Builder packBuilder : topicPackMap.values()) {
senderMsgPackBuilder.addMessagePack(packBuilder);
}
requestBuilder.addMessages(senderMsgPackBuilder.build());
- });
+ itr.remove();
+ }
DistRequest request = requestBuilder.build();
return AsyncRetry.exec(() -> execute(request), retryTimeoutNanos);
}
diff --git a/bifromq-dist/bifromq-dist-server/src/main/java/org/apache/bifromq/dist/server/scheduler/BatchDistServerCall.java b/bifromq-dist/bifromq-dist-server/src/main/java/org/apache/bifromq/dist/server/scheduler/BatchDistServerCall.java
index 25f84bf6..6cb199b0 100644
--- a/bifromq-dist/bifromq-dist-server/src/main/java/org/apache/bifromq/dist/server/scheduler/BatchDistServerCall.java
+++ b/bifromq-dist/bifromq-dist-server/src/main/java/org/apache/bifromq/dist/server/scheduler/BatchDistServerCall.java
@@ -74,7 +74,7 @@ class BatchDistServerCall implements IBatchCall<TenantPubRequest, DistServerCall
private final String orderKey;
private final Queue<ICallTask<TenantPubRequest, DistServerCallResult, DistServerCallBatcherKey>> tasks =
new ArrayDeque<>();
- private Map<String, Map<ClientInfo, Iterable<Message>>> batch = new HashMap<>(128);
+ private final Map<String, Map<ClientInfo, Iterable<Message>>> batch = new HashMap<>(128);
BatchDistServerCall(IBaseKVStoreClient distWorkerClient, DistServerCallBatcherKey batcherKey) {
this.distWorkerClient = distWorkerClient;
@@ -99,7 +99,7 @@ class BatchDistServerCall implements IBatchCall<TenantPubRequest, DistServerCall
@Override
public void reset() {
- batch = new HashMap<>(128);
+ batch.clear();
}
@Override
diff --git a/bifromq-dist/bifromq-dist-server/src/test/java/org/apache/bifromq/dist/server/DistServiceTest.java b/bifromq-dist/bifromq-dist-server/src/test/java/org/apache/bifromq/dist/server/DistServiceTest.java
index 8a3cd46a..edb96f62 100644
--- a/bifromq-dist/bifromq-dist-server/src/test/java/org/apache/bifromq/dist/server/DistServiceTest.java
+++ b/bifromq-dist/bifromq-dist-server/src/test/java/org/apache/bifromq/dist/server/DistServiceTest.java
@@ -24,6 +24,10 @@ import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.when;
+import java.time.Duration;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import lombok.extern.slf4j.Slf4j;
import org.apache.bifromq.basecluster.AgentHostOptions;
import org.apache.bifromq.basecluster.IAgentHost;
import org.apache.bifromq.basecrdt.service.CRDTServiceOptions;
@@ -41,16 +45,12 @@ import org.apache.bifromq.dist.client.IDistClient;
import org.apache.bifromq.dist.worker.IDistWorker;
import org.apache.bifromq.plugin.eventcollector.Event;
import org.apache.bifromq.plugin.eventcollector.IEventCollector;
+import org.apache.bifromq.plugin.resourcethrottler.IResourceThrottler;
import org.apache.bifromq.plugin.settingprovider.ISettingProvider;
import org.apache.bifromq.plugin.settingprovider.Setting;
import org.apache.bifromq.plugin.subbroker.IDeliverer;
import org.apache.bifromq.plugin.subbroker.ISubBroker;
import org.apache.bifromq.plugin.subbroker.ISubBrokerManager;
-import org.apache.bifromq.plugin.resourcethrottler.IResourceThrottler;
-import java.time.Duration;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import lombok.extern.slf4j.Slf4j;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
import org.testng.annotations.AfterClass;
@@ -134,6 +134,7 @@ public abstract class DistServiceTest {
.bgTaskExecutor(bgTaskExecutor)
.storeOptions(kvRangeStoreOptions)
.subBrokerManager(subBrokerMgr)
+ .settingProvider(settingProvider)
.build();
distServer = IDistServer.builder()
.rpcServerBuilder(rpcServerBuilder)
diff --git a/bifromq-dist/bifromq-dist-worker-schema/src/main/java/org/apache/bifromq/dist/worker/schema/GroupMatching.java b/bifromq-dist/bifromq-dist-worker-schema/src/main/java/org/apache/bifromq/dist/worker/schema/GroupMatching.java
index b20438c6..ff745eee 100644
--- a/bifromq-dist/bifromq-dist-worker-schema/src/main/java/org/apache/bifromq/dist/worker/schema/GroupMatching.java
+++ b/bifromq-dist/bifromq-dist-worker-schema/src/main/java/org/apache/bifromq/dist/worker/schema/GroupMatching.java
@@ -14,17 +14,17 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
package org.apache.bifromq.dist.worker.schema;
-import org.apache.bifromq.type.RouteMatcher;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import lombok.EqualsAndHashCode;
import lombok.ToString;
+import org.apache.bifromq.type.RouteMatcher;
/**
* Represent a group matching route.
@@ -38,7 +38,6 @@ public final class GroupMatching extends Matching {
public final List<NormalMatching> receiverList;
private final Map<String, Long> receivers;
-
GroupMatching(String tenantId, RouteMatcher matcher, Map<String, Long> members) {
super(tenantId, matcher);
assert matcher.getType() != RouteMatcher.Type.Normal;
diff --git a/bifromq-dist/bifromq-dist-worker-schema/src/main/java/org/apache/bifromq/dist/worker/schema/Matching.java b/bifromq-dist/bifromq-dist-worker-schema/src/main/java/org/apache/bifromq/dist/worker/schema/Matching.java
index 0a2c20f1..60b8d9aa 100644
--- a/bifromq-dist/bifromq-dist-worker-schema/src/main/java/org/apache/bifromq/dist/worker/schema/Matching.java
+++ b/bifromq-dist/bifromq-dist-worker-schema/src/main/java/org/apache/bifromq/dist/worker/schema/Matching.java
@@ -14,14 +14,14 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
package org.apache.bifromq.dist.worker.schema;
-import org.apache.bifromq.type.RouteMatcher;
import lombok.EqualsAndHashCode;
import lombok.ToString;
+import org.apache.bifromq.type.RouteMatcher;
/**
* The abstract class of matching route.
diff --git a/bifromq-dist/bifromq-dist-worker-schema/src/main/java/org/apache/bifromq/dist/worker/schema/NormalMatching.java b/bifromq-dist/bifromq-dist-worker-schema/src/main/java/org/apache/bifromq/dist/worker/schema/NormalMatching.java
index 5a17a247..07b95f22 100644
--- a/bifromq-dist/bifromq-dist-worker-schema/src/main/java/org/apache/bifromq/dist/worker/schema/NormalMatching.java
+++ b/bifromq-dist/bifromq-dist-worker-schema/src/main/java/org/apache/bifromq/dist/worker/schema/NormalMatching.java
@@ -14,15 +14,15 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
package org.apache.bifromq.dist.worker.schema;
-import org.apache.bifromq.type.MatchInfo;
-import org.apache.bifromq.type.RouteMatcher;
import lombok.EqualsAndHashCode;
import lombok.ToString;
+import org.apache.bifromq.type.MatchInfo;
+import org.apache.bifromq.type.RouteMatcher;
/**
* Represent a normal matching route.
diff --git a/bifromq-dist/bifromq-dist-worker/pom.xml b/bifromq-dist/bifromq-dist-worker/pom.xml
index c4b4e503..bd77847c 100644
--- a/bifromq-dist/bifromq-dist-worker/pom.xml
+++ b/bifromq-dist/bifromq-dist-worker/pom.xml
@@ -78,6 +78,10 @@
<groupId>org.apache.bifromq</groupId>
<artifactId>bifromq-plugin-event-collector-helper</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.bifromq</groupId>
+ <artifactId>bifromq-plugin-setting-provider</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.bifromq</groupId>
<artifactId>bifromq-dist-client</artifactId>
diff --git a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/DeliverExecutorGroup.java b/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/DeliverExecutorGroup.java
index e6fe2173..e0f87f39 100644
--- a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/DeliverExecutorGroup.java
+++ b/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/DeliverExecutorGroup.java
@@ -14,7 +14,7 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
package org.apache.bifromq.dist.worker;
@@ -22,7 +22,23 @@ package org.apache.bifromq.dist.worker;
import static org.apache.bifromq.metrics.TenantMetric.MqttPersistentFanOutBytes;
import static org.apache.bifromq.plugin.eventcollector.ThreadLocalEventPool.getLocal;
import static org.apache.bifromq.plugin.resourcethrottler.TenantResourceType.TotalPersistentFanOutBytesPerSeconds;
+import static org.apache.bifromq.plugin.resourcethrottler.TenantResourceType.TotalTransientFanOutBytesPerSeconds;
+import static org.apache.bifromq.plugin.settingprovider.Setting.MaxGroupFanout;
+import static org.apache.bifromq.plugin.settingprovider.Setting.MaxPersistentFanout;
+import static org.apache.bifromq.plugin.settingprovider.Setting.MaxPersistentFanoutBytes;
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.LoadingCache;
+import com.github.benmanes.caffeine.cache.RemovalListener;
+import com.github.benmanes.caffeine.cache.Scheduler;
+import com.google.common.base.Charsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
import org.apache.bifromq.deliverer.IMessageDeliverer;
import org.apache.bifromq.deliverer.TopicMessagePackHolder;
import org.apache.bifromq.dist.worker.schema.GroupMatching;
@@ -31,44 +47,40 @@ import org.apache.bifromq.dist.worker.schema.NormalMatching;
import org.apache.bifromq.metrics.ITenantMeter;
import org.apache.bifromq.plugin.eventcollector.IEventCollector;
import org.apache.bifromq.plugin.eventcollector.OutOfTenantResource;
-import org.apache.bifromq.sysprops.props.DistInlineFanOutThreshold;
+import org.apache.bifromq.plugin.eventcollector.distservice.GroupFanoutThrottled;
+import org.apache.bifromq.plugin.eventcollector.distservice.PersistentFanoutBytesThrottled;
+import org.apache.bifromq.plugin.eventcollector.distservice.PersistentFanoutThrottled;
+import org.apache.bifromq.plugin.resourcethrottler.IResourceThrottler;
+import org.apache.bifromq.plugin.resourcethrottler.TenantResourceType;
+import org.apache.bifromq.plugin.settingprovider.ISettingProvider;
import org.apache.bifromq.sysprops.props.DistTopicMatchExpirySeconds;
import org.apache.bifromq.type.ClientInfo;
import org.apache.bifromq.type.RouteMatcher;
import org.apache.bifromq.type.TopicMessagePack;
import org.apache.bifromq.util.SizeUtil;
-import org.apache.bifromq.plugin.resourcethrottler.IResourceThrottler;
-import org.apache.bifromq.plugin.resourcethrottler.TenantResourceType;
-import com.github.benmanes.caffeine.cache.Cache;
-import com.github.benmanes.caffeine.cache.Caffeine;
-import com.github.benmanes.caffeine.cache.LoadingCache;
-import com.github.benmanes.caffeine.cache.RemovalListener;
-import com.github.benmanes.caffeine.cache.Scheduler;
-import com.google.common.base.Charsets;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ThreadLocalRandom;
-import java.util.concurrent.TimeUnit;
-import lombok.extern.slf4j.Slf4j;
@Slf4j
class DeliverExecutorGroup implements IDeliverExecutorGroup {
// OuterCacheKey: OrderedSharedMatchingKey(<tenantId>, <mqttTopicFilter>)
// InnerCacheKey: ClientInfo(<tenantId>, <type>, <metadata>)
private final LoadingCache<OrderedSharedMatchingKey, Cache<ClientInfo, NormalMatching>> orderedSharedMatching;
- private final int inlineFanOutThreshold = DistInlineFanOutThreshold.INSTANCE.get();
+ private final int inlineFanOutThreshold;
private final IEventCollector eventCollector;
private final IResourceThrottler resourceThrottler;
+ private final ISettingProvider settingProvider;
private final DeliverExecutor[] fanoutExecutors;
DeliverExecutorGroup(IMessageDeliverer deliverer,
IEventCollector eventCollector,
IResourceThrottler resourceThrottler,
- int groupSize) {
- int expirySec = DistTopicMatchExpirySeconds.INSTANCE.get();
+ ISettingProvider settingProvider,
+ int fanoutParallelism,
+ int inlineFanOutThreshold) {
this.eventCollector = eventCollector;
this.resourceThrottler = resourceThrottler;
+ this.settingProvider = settingProvider;
+ this.inlineFanOutThreshold = inlineFanOutThreshold;
+ int expirySec = DistTopicMatchExpirySeconds.INSTANCE.get();
orderedSharedMatching = Caffeine.newBuilder()
.expireAfterAccess(expirySec * 2L, TimeUnit.SECONDS)
.scheduler(Scheduler.systemScheduler())
@@ -81,8 +93,8 @@ class DeliverExecutorGroup implements IDeliverExecutorGroup {
.build(k -> Caffeine.newBuilder()
.expireAfterAccess(expirySec, TimeUnit.SECONDS)
.build());
- fanoutExecutors = new DeliverExecutor[groupSize];
- for (int i = 0; i < groupSize; i++) {
+ fanoutExecutors = new DeliverExecutor[fanoutParallelism];
+ for (int i = 0; i < fanoutParallelism; i++) {
fanoutExecutors[i] = new DeliverExecutor(i, deliverer, eventCollector);
}
}
@@ -101,59 +113,122 @@ class DeliverExecutorGroup implements IDeliverExecutorGroup {
TopicMessagePackHolder messagePackHolder = TopicMessagePackHolder.hold(msgPack);
if (routes.size() == 1) {
Matching matching = routes.iterator().next();
- prepareSend(matching, messagePackHolder, true);
- if (isSendToInbox(matching)) {
- ITenantMeter.get(matching.tenantId()).recordSummary(MqttPersistentFanOutBytes, msgPackSize);
+ switch (matching.type()) {
+ case Normal -> {
+ NormalMatching normalMatching = (NormalMatching) matching;
+ send(normalMatching, messagePackHolder, true);
+ if (normalMatching.subBrokerId() == 1) {
+ ITenantMeter.get(matching.tenantId()).recordSummary(MqttPersistentFanOutBytes, msgPackSize);
+ }
+ }
+ case Group -> send((GroupMatching) matching, messagePackHolder, true);
+ default -> {
+ // never happen
+ }
}
} else if (routes.size() > 1) {
- boolean inline = routes.size() > inlineFanOutThreshold;
- boolean hasTFanOutBandwidth =
- resourceThrottler.hasResource(tenantId, TenantResourceType.TotalTransientFanOutBytesPerSeconds);
- boolean hasTFannedOutUnderThrottled = false;
- boolean hasPFanOutBandwidth =
+ int maxPFanoutCount = settingProvider.provide(MaxPersistentFanout, tenantId);
+ long maxPFanoutBytes = settingProvider.provide(MaxPersistentFanoutBytes, tenantId);
+ int maxGFanoutCount = settingProvider.provide(MaxGroupFanout, tenantId);
+ boolean inline = routes.size() < inlineFanOutThreshold;
+ boolean hasTransientFanoutBandwidth =
+ resourceThrottler.hasResource(tenantId, TotalTransientFanOutBytesPerSeconds);
+ boolean isTransientFanoutThrottled = false;
+ boolean hasPersistentFanoutBandwidth =
resourceThrottler.hasResource(tenantId, TenantResourceType.TotalPersistentFanOutBytesPerSeconds);
- boolean hasPFannedOutUnderThrottled = false;
+ boolean isPersistentFanoutThrottled = false;
+ boolean isGroupFanoutThrottled = false;
// we meter persistent fanout bytes here, since for transient fanout is actually happened in the broker
- long pFanoutBytes = 0;
+ int persistentFanoutCount = 0;
+ long persistentFanoutBytes = 0;
+ int groupFanoutCount = 0;
for (Matching matching : routes) {
- if (isSendToInbox(matching)) {
- if (hasPFanOutBandwidth || !hasPFannedOutUnderThrottled) {
- pFanoutBytes += msgPackSize;
- prepareSend(matching, messagePackHolder, inline);
- if (!hasPFanOutBandwidth) {
- hasPFannedOutUnderThrottled = true;
- for (TopicMessagePack.PublisherPack publisherPack : msgPack.getMessageList()) {
- eventCollector.report(getLocal(OutOfTenantResource.class)
- .reason(TotalPersistentFanOutBytesPerSeconds.name())
- .clientInfo(publisherPack.getPublisher())
- );
+ switch (matching.type()) {
+ case Normal -> {
+ NormalMatching normalMatching = (NormalMatching) matching;
+ if (normalMatching.subBrokerId() == 1) {
+ // persistent fanout
+ if (persistentFanoutCount < maxPFanoutCount && persistentFanoutBytes < maxPFanoutBytes) {
+ if (hasPersistentFanoutBandwidth) {
+ persistentFanoutCount++;
+ persistentFanoutBytes += msgPackSize;
+ send(normalMatching, messagePackHolder, inline);
+ } else {
+ if (!isPersistentFanoutThrottled) {
+ isPersistentFanoutThrottled = true;
+ for (TopicMessagePack.PublisherPack publisherPack : msgPack.getMessageList()) {
+ eventCollector.report(getLocal(OutOfTenantResource.class)
+ .reason(TotalPersistentFanOutBytesPerSeconds.name())
+ .clientInfo(publisherPack.getPublisher())
+ );
+ }
+ }
+ }
+ } else {
+ if (!isPersistentFanoutThrottled) {
+ isPersistentFanoutThrottled = true;
+ // persistent fanout throttled
+ if (persistentFanoutCount >= maxPFanoutCount) {
+ eventCollector.report(getLocal(PersistentFanoutThrottled.class)
+ .tenantId(tenantId)
+ .topic(msgPack.getTopic())
+ .maxCount(maxPFanoutCount)
+ );
+ }
+ if (persistentFanoutBytes >= maxPFanoutBytes) {
+ eventCollector.report(getLocal(PersistentFanoutBytesThrottled.class)
+ .tenantId(tenantId)
+ .topic(msgPack.getTopic())
+ .maxBytes(maxPFanoutBytes)
+ );
+ }
+ }
+ }
+ } else {
+ // transient fanout
+ if (hasTransientFanoutBandwidth) {
+ send(normalMatching, messagePackHolder, inline);
+ } else {
+ if (!isTransientFanoutThrottled) {
+ isTransientFanoutThrottled = true;
+ for (TopicMessagePack.PublisherPack publisherPack : msgPack.getMessageList()) {
+ eventCollector.report(getLocal(OutOfTenantResource.class)
+ .reason(TotalTransientFanOutBytesPerSeconds.name())
+ .clientInfo(publisherPack.getPublisher())
+ );
+ }
+ }
}
}
}
- } else if (hasTFanOutBandwidth || !hasTFannedOutUnderThrottled) {
- prepareSend(matching, messagePackHolder, inline);
- if (!hasTFanOutBandwidth) {
- hasTFannedOutUnderThrottled = true;
- for (TopicMessagePack.PublisherPack publisherPack : msgPack.getMessageList()) {
- eventCollector.report(getLocal(OutOfTenantResource.class)
- .reason(TenantResourceType.TotalTransientFanOutBytesPerSeconds.name())
- .clientInfo(publisherPack.getPublisher())
- );
+ case Group -> {
+ if (groupFanoutCount < maxGFanoutCount) {
+ groupFanoutCount++;
+ send((GroupMatching) matching, messagePackHolder, inline);
+ } else {
+ if (!isGroupFanoutThrottled) {
+ isGroupFanoutThrottled = true;
+ // group fanout throttled
+ eventCollector.report(getLocal(GroupFanoutThrottled.class)
+ .tenantId(tenantId)
+ .topic(msgPack.getTopic())
+ .maxCount(maxGFanoutCount)
+ );
+ }
}
}
+ default -> {
+ // never happen
+ }
}
- if (hasPFannedOutUnderThrottled && hasTFannedOutUnderThrottled) {
+ if (isPersistentFanoutThrottled && isTransientFanoutThrottled && isGroupFanoutThrottled) {
break;
}
}
- ITenantMeter.get(tenantId).recordSummary(MqttPersistentFanOutBytes, pFanoutBytes);
+ ITenantMeter.get(tenantId).recordSummary(MqttPersistentFanOutBytes, persistentFanoutBytes);
}
}
- private boolean isSendToInbox(Matching matching) {
- return matching.type() == Matching.Type.Normal && ((NormalMatching) matching).subBrokerId() == 1;
- }
-
@Override
public void refreshOrderedShareSubRoutes(String tenantId, RouteMatcher routeMatcher) {
if (log.isDebugEnabled()) {
@@ -163,48 +238,41 @@ class DeliverExecutorGroup implements IDeliverExecutorGroup {
new OrderedSharedMatchingKey(tenantId, routeMatcher.getMqttTopicFilter()));
}
- private void prepareSend(Matching matching, TopicMessagePackHolder msgPackHolder, boolean inline) {
- switch (matching.type()) {
- case Normal -> send((NormalMatching) matching, msgPackHolder, inline);
- case Group -> {
- GroupMatching groupMatching = (GroupMatching) matching;
- if (!groupMatching.ordered) {
- // pick one route randomly
- send(groupMatching.receiverList.get(
- ThreadLocalRandom.current().nextInt(groupMatching.receiverList.size())), msgPackHolder, inline);
- } else {
- // ordered shared subscription
- Map<NormalMatching, TopicMessagePack.Builder> orderedRoutes = new HashMap<>();
- for (TopicMessagePack.PublisherPack publisherPack : msgPackHolder.messagePack.getMessageList()) {
- ClientInfo sender = publisherPack.getPublisher();
- NormalMatching matchedInbox = orderedSharedMatching
- .get(new OrderedSharedMatchingKey(groupMatching.tenantId(),
- groupMatching.matcher.getMqttTopicFilter()))
- .get(sender, senderInfo -> {
- RendezvousHash<ClientInfo, NormalMatching> hash =
- RendezvousHash.<ClientInfo, NormalMatching>builder()
- .keyFunnel((from, into) -> into.putInt(from.hashCode()))
- .nodeFunnel((from, into) -> into.putString(from.receiverUrl(), Charsets.UTF_8))
- .nodes(groupMatching.receiverList)
- .build();
- NormalMatching matchRecord = hash.get(senderInfo);
- log.debug(
- "Ordered shared matching: sender={}: topicFilter={}, receiverId={}, subBroker={}",
- senderInfo,
- matchRecord.mqttTopicFilter(),
- matchRecord.matchInfo().getReceiverId(),
- matchRecord.subBrokerId());
- return matchRecord;
- });
- // ordered share sub
- orderedRoutes.computeIfAbsent(matchedInbox, k -> TopicMessagePack.newBuilder())
- .setTopic(msgPackHolder.messagePack.getTopic())
- .addMessage(publisherPack);
- }
- orderedRoutes.forEach((route, msgPackBuilder) ->
- send(route, TopicMessagePackHolder.hold(msgPackBuilder.build()), inline));
- }
+ private void send(GroupMatching groupMatching, TopicMessagePackHolder msgPackHolder, boolean inline) {
+ if (!groupMatching.ordered) {
+ // pick one route randomly
+ send(groupMatching.receiverList.get(
+ ThreadLocalRandom.current().nextInt(groupMatching.receiverList.size())), msgPackHolder, inline);
+ } else {
+ // ordered shared subscription
+ Map<NormalMatching, TopicMessagePack.Builder> orderedRoutes = new HashMap<>();
+ for (TopicMessagePack.PublisherPack publisherPack : msgPackHolder.messagePack.getMessageList()) {
+ ClientInfo sender = publisherPack.getPublisher();
+ NormalMatching matchedInbox = orderedSharedMatching
+ .get(new OrderedSharedMatchingKey(groupMatching.tenantId(),
+ groupMatching.matcher.getMqttTopicFilter()))
+ .get(sender, senderInfo -> {
+ RendezvousHash<ClientInfo, NormalMatching> hash =
+ RendezvousHash.<ClientInfo, NormalMatching>builder()
+ .keyFunnel((from, into) -> into.putInt(from.hashCode()))
+ .nodeFunnel((from, into) -> into.putString(from.receiverUrl(), Charsets.UTF_8))
+ .nodes(groupMatching.receiverList)
+ .build();
+ NormalMatching matchRecord = hash.get(senderInfo);
+ log.debug("Ordered shared matching: sender={}: topicFilter={}, receiverId={}, subBroker={}",
+ senderInfo,
+ matchRecord.mqttTopicFilter(),
+ matchRecord.matchInfo().getReceiverId(),
+ matchRecord.subBrokerId());
+ return matchRecord;
+ });
+ // ordered share sub
+ orderedRoutes.computeIfAbsent(matchedInbox, k -> TopicMessagePack.newBuilder())
+ .setTopic(msgPackHolder.messagePack.getTopic())
+ .addMessage(publisherPack);
}
+ orderedRoutes.forEach((route, msgPackBuilder) ->
+ send(route, TopicMessagePackHolder.hold(msgPackBuilder.build()), inline));
}
}
diff --git a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/DistWorker.java b/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/DistWorker.java
index ed75a9ef..6ea8315a 100644
--- a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/DistWorker.java
+++ b/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/DistWorker.java
@@ -69,7 +69,10 @@ class DistWorker implements IDistWorker {
builder.resourceThrottler,
builder.subBrokerManager,
this.messageDeliverer,
- builder.loadEstimateWindow);
+ builder.settingProvider,
+ builder.loadEstimateWindow,
+ builder.fanoutParallelism,
+ builder.inlineFanoutThreshold);
Map<String, IDistWorkerBalancerFactory> loadedFactories = BaseHookLoader.load(IDistWorkerBalancerFactory.class);
for (String factoryName : builder.balancerFactoryConfig.keySet()) {
if (!loadedFactories.containsKey(factoryName)) {
diff --git a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/DistWorkerBuilder.java b/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/DistWorkerBuilder.java
index 1968e39f..a52585fd 100644
--- a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/DistWorkerBuilder.java
+++ b/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/DistWorkerBuilder.java
@@ -14,20 +14,11 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
package org.apache.bifromq.dist.worker;
-import org.apache.bifromq.basecluster.IAgentHost;
-import org.apache.bifromq.basekv.client.IBaseKVStoreClient;
-import org.apache.bifromq.basekv.metaservice.IBaseKVMetaService;
-import org.apache.bifromq.basekv.store.option.KVRangeStoreOptions;
-import org.apache.bifromq.baserpc.server.RPCServerBuilder;
-import org.apache.bifromq.dist.client.IDistClient;
-import org.apache.bifromq.plugin.eventcollector.IEventCollector;
-import org.apache.bifromq.plugin.subbroker.ISubBrokerManager;
-import org.apache.bifromq.plugin.resourcethrottler.IResourceThrottler;
import com.google.protobuf.Struct;
import java.time.Duration;
import java.util.HashMap;
@@ -37,6 +28,18 @@ import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import lombok.Setter;
import lombok.experimental.Accessors;
+import org.apache.bifromq.basecluster.IAgentHost;
+import org.apache.bifromq.basekv.client.IBaseKVStoreClient;
+import org.apache.bifromq.basekv.metaservice.IBaseKVMetaService;
+import org.apache.bifromq.basekv.store.option.KVRangeStoreOptions;
+import org.apache.bifromq.baserpc.server.RPCServerBuilder;
+import org.apache.bifromq.dist.client.IDistClient;
+import org.apache.bifromq.plugin.eventcollector.IEventCollector;
+import org.apache.bifromq.plugin.resourcethrottler.IResourceThrottler;
+import org.apache.bifromq.plugin.settingprovider.ISettingProvider;
+import org.apache.bifromq.plugin.subbroker.ISubBrokerManager;
+import org.apache.bifromq.sysprops.props.DistFanOutParallelism;
+import org.apache.bifromq.sysprops.props.DistInlineFanOutThreshold;
/**
* The builder for building Dist Worker.
@@ -57,6 +60,7 @@ public class DistWorkerBuilder {
IDistClient distClient;
IBaseKVStoreClient distWorkerClient;
ISubBrokerManager subBrokerManager;
+ ISettingProvider settingProvider;
KVRangeStoreOptions storeOptions;
Duration gcInterval = Duration.ofMinutes(5);
Duration bootstrapDelay = Duration.ofSeconds(15);
@@ -64,6 +68,8 @@ public class DistWorkerBuilder {
Duration balancerRetryDelay = Duration.ofSeconds(5);
Map<String, Struct> balancerFactoryConfig = new HashMap<>();
Duration loadEstimateWindow = Duration.ofSeconds(5);
+ int fanoutParallelism = DistFanOutParallelism.INSTANCE.get();
+ int inlineFanoutThreshold = DistInlineFanOutThreshold.INSTANCE.get();
Map<String, String> attributes = new HashMap<>();
public IDistWorker build() {
diff --git a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/DistWorkerCoProcFactory.java b/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/DistWorkerCoProcFactory.java
index 8b69c7e3..4b0b6c52 100644
--- a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/DistWorkerCoProcFactory.java
+++ b/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/DistWorkerCoProcFactory.java
@@ -14,11 +14,24 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
package org.apache.bifromq.dist.worker;
+import com.google.common.util.concurrent.MoreExecutors;
+import io.micrometer.core.instrument.Metrics;
+import io.micrometer.core.instrument.binder.jvm.ExecutorServiceMetrics;
+import java.time.Duration;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.ForkJoinWorkerThread;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
+import lombok.extern.slf4j.Slf4j;
import org.apache.bifromq.basekv.proto.KVRangeId;
import org.apache.bifromq.basekv.store.api.IKVCloseableReader;
import org.apache.bifromq.basekv.store.api.IKVRangeCoProc;
@@ -32,33 +45,23 @@ import org.apache.bifromq.dist.worker.cache.ISubscriptionCache;
import org.apache.bifromq.dist.worker.cache.SubscriptionCache;
import org.apache.bifromq.dist.worker.hinter.FanoutSplitHinter;
import org.apache.bifromq.plugin.eventcollector.IEventCollector;
+import org.apache.bifromq.plugin.resourcethrottler.IResourceThrottler;
+import org.apache.bifromq.plugin.settingprovider.ISettingProvider;
import org.apache.bifromq.plugin.subbroker.ISubBrokerManager;
-import org.apache.bifromq.sysprops.props.DistFanOutParallelism;
import org.apache.bifromq.sysprops.props.DistMatchParallelism;
import org.apache.bifromq.sysprops.props.DistWorkerFanOutSplitThreshold;
-import org.apache.bifromq.plugin.resourcethrottler.IResourceThrottler;
-import com.google.common.util.concurrent.MoreExecutors;
-import io.micrometer.core.instrument.Metrics;
-import io.micrometer.core.instrument.binder.jvm.ExecutorServiceMetrics;
-import java.time.Duration;
-import java.util.List;
-import java.util.Optional;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.ForkJoinPool;
-import java.util.concurrent.ForkJoinWorkerThread;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.function.Supplier;
-import lombok.extern.slf4j.Slf4j;
@Slf4j
public class DistWorkerCoProcFactory implements IKVRangeCoProcFactory {
private final IEventCollector eventCollector;
private final IResourceThrottler resourceThrottler;
private final IMessageDeliverer deliverer;
+ private final ISettingProvider settingProvider;
private final ISubscriptionCleaner subscriptionChecker;
private final ExecutorService matchExecutor;
private final Duration loadEstWindow;
+ private final int fanoutParallelism;
+ private final int inlineFanOutThreshold;
private final int fanoutSplitThreshold = DistWorkerFanOutSplitThreshold.INSTANCE.get();
public DistWorkerCoProcFactory(IDistClient distClient,
@@ -66,11 +69,17 @@ public class DistWorkerCoProcFactory implements IKVRangeCoProcFactory {
IResourceThrottler resourceThrottler,
ISubBrokerManager subBrokerManager,
IMessageDeliverer messageDeliverer,
- Duration loadEstimateWindow) {
+ ISettingProvider settingProvider,
+ Duration loadEstimateWindow,
+ int fanoutParallelism,
+ int inlineFanOutThreshold) {
this.eventCollector = eventCollector;
this.resourceThrottler = resourceThrottler;
this.loadEstWindow = loadEstimateWindow;
this.deliverer = messageDeliverer;
+ this.settingProvider = settingProvider;
+ this.fanoutParallelism = fanoutParallelism;
+ this.inlineFanOutThreshold = inlineFanOutThreshold;
subscriptionChecker = new SubscriptionCleaner(subBrokerManager, distClient);
matchExecutor = ExecutorServiceMetrics.monitor(Metrics.globalRegistry,
@@ -100,12 +109,13 @@ public class DistWorkerCoProcFactory implements IKVRangeCoProcFactory {
@Override
public IKVRangeCoProc createCoProc(String clusterId, String storeId, KVRangeId id,
Supplier<IKVCloseableReader> rangeReaderProvider) {
- ISubscriptionCache routeCache = new SubscriptionCache(id, rangeReaderProvider, matchExecutor);
+ ISubscriptionCache routeCache = new SubscriptionCache(id, rangeReaderProvider,
+ settingProvider, eventCollector, matchExecutor);
ITenantsState tenantsState = new TenantsState(rangeReaderProvider.get(),
"clusterId", clusterId, "storeId", storeId, "rangeId", KVRangeIdUtil.toString(id));
IDeliverExecutorGroup deliverExecutorGroup = new DeliverExecutorGroup(
- deliverer, eventCollector, resourceThrottler, DistFanOutParallelism.INSTANCE.get());
+ deliverer, eventCollector, resourceThrottler, settingProvider, fanoutParallelism, inlineFanOutThreshold);
return new DistWorkerCoProc(
id, rangeReaderProvider, routeCache, tenantsState, deliverExecutorGroup, subscriptionChecker);
}
diff --git a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/ITenantRouteMatcher.java b/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/IMatchedRoutes.java
similarity index 52%
copy from bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/ITenantRouteMatcher.java
copy to bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/IMatchedRoutes.java
index 191155d9..0a1756da 100644
--- a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/ITenantRouteMatcher.java
+++ b/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/IMatchedRoutes.java
@@ -14,24 +14,51 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
package org.apache.bifromq.dist.worker.cache;
-import org.apache.bifromq.dist.worker.schema.Matching;
-import java.util.Map;
import java.util.Set;
+import org.apache.bifromq.dist.worker.schema.Matching;
/**
- * The interface for tenant route matcher.
+ * The result of matching a topic within a boundary.
*/
-public interface ITenantRouteMatcher {
+public interface IMatchedRoutes {
+ /**
+ * The maximum persistent fanout count for the matched topic.
+ *
+ * @return the maximum count of persistent fanout matches
+ */
+ int maxPersistentFanout();
+
+ /**
+ * The maximum group fanout count for the matched topic.
+ *
+ * @return the maximum count of group fanout matches
+ */
+ int maxGroupFanout();
+
/**
- * Match the topic within the given boundary.
+ * The persistent fanout count for the matched topic.
*
- * @param topics the topics to match
- * @return the result
+ * @return the count of persistent fanout matches
*/
- Map<String, Set<Matching>> matchAll(Set<String> topics);
+ int persistentFanout();
+
+ /**
+ * The count of group fanout for the matched topic.
+ *
+ * @return the count of group fanout
+ */
+ int groupFanout();
+
+ /**
+ * The matched routes.
+ *
+ * @return the matched routes.
+ */
+ Set<Matching> routes();
+
}
diff --git a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/ITenantRouteMatcher.java b/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/ITenantRouteMatcher.java
index 191155d9..9690ce03 100644
--- a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/ITenantRouteMatcher.java
+++ b/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/ITenantRouteMatcher.java
@@ -14,12 +14,11 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
package org.apache.bifromq.dist.worker.cache;
-import org.apache.bifromq.dist.worker.schema.Matching;
import java.util.Map;
import java.util.Set;
@@ -31,7 +30,9 @@ public interface ITenantRouteMatcher {
* Match the topic within the given boundary.
*
* @param topics the topics to match
+ * @param maxPersistentFanoutCount the maximum number of persistent fanout matches to return
+ * @param maxGroupFanoutCount the maximum number of group fanout matches to return
* @return the result
*/
- Map<String, Set<Matching>> matchAll(Set<String> topics);
+ Map<String, IMatchedRoutes> matchAll(Set<String> topics, int maxPersistentFanoutCount, int maxGroupFanoutCount);
}
diff --git a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/SubscriptionCache.java b/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/SubscriptionCache.java
index 99073abf..5ff3804d 100644
--- a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/SubscriptionCache.java
+++ b/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/SubscriptionCache.java
@@ -14,7 +14,7 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
package org.apache.bifromq.dist.worker.cache;
@@ -26,13 +26,6 @@ import static org.apache.bifromq.dist.worker.cache.SubscriptionCache.TenantKey.n
import static org.apache.bifromq.dist.worker.cache.SubscriptionCache.TenantKey.refreshExpiry;
import static org.apache.bifromq.dist.worker.schema.KVSchemaUtil.tenantBeginKey;
-import org.apache.bifromq.basekv.proto.Boundary;
-import org.apache.bifromq.basekv.proto.KVRangeId;
-import org.apache.bifromq.basekv.store.api.IKVCloseableReader;
-import org.apache.bifromq.basekv.utils.KVRangeIdUtil;
-import org.apache.bifromq.dist.worker.schema.Matching;
-import org.apache.bifromq.sysprops.props.DistTopicMatchExpirySeconds;
-import org.apache.bifromq.type.RouteMatcher;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.Expiry;
import com.github.benmanes.caffeine.cache.LoadingCache;
@@ -49,6 +42,16 @@ import java.util.concurrent.Executor;
import java.util.function.Supplier;
import lombok.EqualsAndHashCode;
import lombok.extern.slf4j.Slf4j;
+import org.apache.bifromq.basekv.proto.Boundary;
+import org.apache.bifromq.basekv.proto.KVRangeId;
+import org.apache.bifromq.basekv.store.api.IKVCloseableReader;
+import org.apache.bifromq.basekv.utils.KVRangeIdUtil;
+import org.apache.bifromq.dist.worker.schema.Matching;
+import org.apache.bifromq.plugin.eventcollector.IEventCollector;
+import org.apache.bifromq.plugin.settingprovider.ISettingProvider;
+import org.apache.bifromq.sysprops.props.DistCachedRoutesFanoutCheckIntervalSeconds;
+import org.apache.bifromq.sysprops.props.DistTopicMatchExpirySeconds;
+import org.apache.bifromq.type.RouteMatcher;
/**
* Cache for subscription matching.
@@ -59,11 +62,15 @@ public class SubscriptionCache implements ISubscriptionCache {
private final LoadingCache<TenantKey, ITenantRouteCache> tenantCache;
private volatile Boundary boundary;
- public SubscriptionCache(KVRangeId id, Supplier<IKVCloseableReader> rangeReaderProvider, Executor matchExecutor) {
- this(id, new TenantRouteCacheFactory(rangeReaderProvider,
- Duration.ofSeconds(DistTopicMatchExpirySeconds.INSTANCE.get()), matchExecutor,
- "id", KVRangeIdUtil.toString(id)),
- Ticker.systemTicker());
+ public SubscriptionCache(KVRangeId id,
+ Supplier<IKVCloseableReader> rangeReaderProvider,
+ ISettingProvider settingProvider,
+ IEventCollector eventCollector,
+ Executor matchExecutor) {
+ this(id, new TenantRouteCacheFactory(rangeReaderProvider, settingProvider, eventCollector,
+ Duration.ofSeconds(DistTopicMatchExpirySeconds.INSTANCE.get()),
+ Duration.ofSeconds(DistCachedRoutesFanoutCheckIntervalSeconds.INSTANCE.get()),
+ matchExecutor, "id", KVRangeIdUtil.toString(id)), Ticker.systemTicker());
}
public SubscriptionCache(KVRangeId id,
diff --git a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/TenantRouteCache.java b/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/TenantRouteCache.java
index d62a1196..e11d4be3 100644
--- a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/TenantRouteCache.java
+++ b/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/TenantRouteCache.java
@@ -14,18 +14,15 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
package org.apache.bifromq.dist.worker.cache;
-import org.apache.bifromq.basekv.proto.Boundary;
-import org.apache.bifromq.dist.worker.TopicIndex;
-import org.apache.bifromq.dist.worker.schema.Matching;
-import org.apache.bifromq.metrics.ITenantMeter;
-import org.apache.bifromq.metrics.TenantMetric;
-import org.apache.bifromq.sysprops.props.DistMaxCachedRoutesPerTenant;
-import org.apache.bifromq.type.RouteMatcher;
+import static java.util.Collections.singleton;
+import static org.apache.bifromq.plugin.settingprovider.Setting.MaxGroupFanout;
+import static org.apache.bifromq.plugin.settingprovider.Setting.MaxPersistentFanout;
+
import com.github.benmanes.caffeine.cache.AsyncLoadingCache;
import com.github.benmanes.caffeine.cache.CacheLoader;
import com.github.benmanes.caffeine.cache.Caffeine;
@@ -33,7 +30,6 @@ import com.github.benmanes.caffeine.cache.Scheduler;
import com.github.benmanes.caffeine.cache.Ticker;
import com.github.benmanes.caffeine.cache.Weigher;
import java.time.Duration;
-import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -41,24 +37,37 @@ import java.util.NavigableSet;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
+import org.apache.bifromq.basekv.proto.Boundary;
+import org.apache.bifromq.dist.worker.TopicIndex;
+import org.apache.bifromq.dist.worker.schema.Matching;
+import org.apache.bifromq.metrics.ITenantMeter;
+import org.apache.bifromq.metrics.TenantMetric;
+import org.apache.bifromq.plugin.settingprovider.ISettingProvider;
+import org.apache.bifromq.sysprops.props.DistMaxCachedRoutesPerTenant;
+import org.apache.bifromq.type.RouteMatcher;
import org.checkerframework.checker.index.qual.NonNegative;
import org.checkerframework.checker.nullness.qual.Nullable;
class TenantRouteCache implements ITenantRouteCache {
private final String tenantId;
- private final AsyncLoadingCache<RouteCacheKey, Set<Matching>> routesCache;
+ private final AsyncLoadingCache<RouteCacheKey, IMatchedRoutes> routesCache;
private final TopicIndex<RouteCacheKey> index;
TenantRouteCache(String tenantId,
ITenantRouteMatcher matcher,
+ ISettingProvider settingProvider,
Duration expiryAfterAccess,
+ Duration fanoutCheckInterval,
Executor matchExecutor) {
- this(tenantId, matcher, expiryAfterAccess, Ticker.systemTicker(), matchExecutor);
+ this(tenantId, matcher, settingProvider, expiryAfterAccess, fanoutCheckInterval, Ticker.systemTicker(),
+ matchExecutor);
}
TenantRouteCache(String tenantId,
ITenantRouteMatcher matcher,
+ ISettingProvider settingProvider,
Duration expiryAfterAccess,
+ Duration fanoutCheckInterval,
Ticker ticker,
Executor matchExecutor) {
this.tenantId = tenantId;
@@ -68,13 +77,14 @@ class TenantRouteCache implements ITenantRouteCache {
.ticker(ticker)
.executor(matchExecutor)
.maximumWeight(DistMaxCachedRoutesPerTenant.INSTANCE.get())
- .weigher(new Weigher<RouteCacheKey, Set<Matching>>() {
+ .weigher(new Weigher<RouteCacheKey, IMatchedRoutes>() {
@Override
- public @NonNegative int weigh(RouteCacheKey key, Set<Matching> value) {
- return value.size();
+ public @NonNegative int weigh(RouteCacheKey key, IMatchedRoutes value) {
+ return value.routes().size();
}
})
.expireAfterAccess(expiryAfterAccess)
+ .refreshAfterWrite(fanoutCheckInterval)
.evictionListener((key, value, cause) -> {
if (key != null) {
index.remove(key.topic, key);
@@ -82,21 +92,27 @@ class TenantRouteCache implements ITenantRouteCache {
})
.buildAsync(new CacheLoader<>() {
@Override
- public @Nullable Set<Matching> load(RouteCacheKey key) {
+ public @Nullable IMatchedRoutes load(RouteCacheKey key) {
+ int maxPersistentFanouts = settingProvider.provide(MaxPersistentFanout, tenantId);
+ int maxGroupFanouts = settingProvider.provide(MaxGroupFanout, tenantId);
ITenantMeter.get(tenantId).recordCount(TenantMetric.MqttRouteCacheMissCount);
- Map<String, Set<Matching>> results = matcher.matchAll(Collections.singleton(key.topic));
+ Map<String, IMatchedRoutes> results = matcher.matchAll(singleton(key.topic),
+ maxPersistentFanouts, maxGroupFanouts);
index.add(key.topic, key);
return results.get(key.topic);
}
@Override
- public Map<RouteCacheKey, Set<Matching>> loadAll(Set<? extends RouteCacheKey> keys) {
+ public Map<RouteCacheKey, IMatchedRoutes> loadAll(Set<? extends RouteCacheKey> keys) {
ITenantMeter.get(tenantId).recordCount(TenantMetric.MqttRouteCacheMissCount, keys.size());
Map<String, RouteCacheKey> topicToKeyMap = new HashMap<>();
keys.forEach(k -> topicToKeyMap.put(k.topic(), k));
- Map<String, Set<Matching>> resultMap = matcher.matchAll(topicToKeyMap.keySet());
- Map<RouteCacheKey, Set<Matching>> result = new HashMap<>();
- for (Map.Entry<String, Set<Matching>> entry : resultMap.entrySet()) {
+ int maxPersistentFanouts = settingProvider.provide(MaxPersistentFanout, tenantId);
+ int maxGroupFanouts = settingProvider.provide(MaxGroupFanout, tenantId);
+ Map<String, IMatchedRoutes> resultMap = matcher.matchAll(topicToKeyMap.keySet(),
+ maxPersistentFanouts, maxGroupFanouts);
+ Map<RouteCacheKey, IMatchedRoutes> result = new HashMap<>();
+ for (Map.Entry<String, IMatchedRoutes> entry : resultMap.entrySet()) {
RouteCacheKey key = topicToKeyMap.get(entry.getKey());
result.put(key, entry.getValue());
index.add(key.topic, key);
@@ -105,14 +121,61 @@ class TenantRouteCache implements ITenantRouteCache {
}
@Override
- public @Nullable Set<Matching> reload(RouteCacheKey key, Set<Matching> oldValue) {
- Map<String, Set<Matching>> results = matcher.matchAll(Collections.singleton(key.topic));
- return results.get(key.topic);
+ public @Nullable IMatchedRoutes reload(RouteCacheKey key, IMatchedRoutes oldValue) {
+ int maxPersistentFanouts = settingProvider.provide(MaxPersistentFanout, tenantId);
+ int maxGroupFanouts = settingProvider.provide(MaxGroupFanout, tenantId);
+ if (needRefresh(oldValue, maxPersistentFanouts, maxGroupFanouts)) {
+ Map<String, IMatchedRoutes> results = matcher.matchAll(singleton(key.topic),
+ maxPersistentFanouts, maxGroupFanouts);
+ return results.get(key.topic);
+ } else if (oldValue.maxPersistentFanout() != maxPersistentFanouts
+ || oldValue.maxGroupFanout() != maxGroupFanouts) {
+ return new IMatchedRoutes() {
+ @Override
+ public int maxPersistentFanout() {
+ return maxPersistentFanouts;
+ }
+
+ @Override
+ public int maxGroupFanout() {
+ return maxGroupFanouts;
+ }
+
+ @Override
+ public int persistentFanout() {
+ return oldValue.persistentFanout();
+ }
+
+ @Override
+ public int groupFanout() {
+ return oldValue.groupFanout();
+ }
+
+ @Override
+ public Set<Matching> routes() {
+ return oldValue.routes();
+ }
+ };
+ } else {
+ return oldValue;
+ }
}
});
ITenantMeter.gauging(tenantId, TenantMetric.MqttRouteCacheSize, routesCache.synchronous()::estimatedSize);
}
+ private boolean needRefresh(IMatchedRoutes cachedRoutes, int maxPersistentFanouts, int maxGroupFanouts) {
+ return needRefresh(maxPersistentFanouts, cachedRoutes.maxPersistentFanout(), cachedRoutes.persistentFanout())
+ || needRefresh(maxGroupFanouts, cachedRoutes.maxGroupFanout(), cachedRoutes.groupFanout());
+ }
+
+ private boolean needRefresh(int currentLimit, int previousLimit, int currentFanout) {
+ // current limit increases and cachedRoutes reached the previous limit, need refresh
+ // current limit decreases and cachedRoutes exceeded the current limit, need refresh
+ return (previousLimit < currentLimit && currentFanout == previousLimit)
+ || (previousLimit > currentLimit && currentFanout > currentLimit);
+ }
+
@Override
public boolean isCached(List<String> filterLevels) {
return !index.match(filterLevels).isEmpty();
@@ -122,6 +185,8 @@ class TenantRouteCache implements ITenantRouteCache {
public void refresh(NavigableSet<RouteMatcher> routeMatchers) {
routeMatchers.forEach(topicFilter -> {
for (RouteCacheKey cacheKey : index.match(topicFilter.getFilterLevelList())) {
+ // we must invalidate the cache entry first to ensure the refresh will trigger a load
+ routesCache.synchronous().invalidate(cacheKey);
routesCache.synchronous().refresh(cacheKey);
}
});
@@ -129,7 +194,7 @@ class TenantRouteCache implements ITenantRouteCache {
@Override
public CompletableFuture<Set<Matching>> getMatch(String topic, Boundary currentTenantRange) {
- return routesCache.get(new RouteCacheKey(topic, currentTenantRange));
+ return routesCache.get(new RouteCacheKey(topic, currentTenantRange)).thenApply(IMatchedRoutes::routes);
}
@Override
diff --git a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/TenantRouteCacheFactory.java b/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/TenantRouteCacheFactory.java
index 2a87ade0..363f94c1 100644
--- a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/TenantRouteCacheFactory.java
+++ b/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/TenantRouteCacheFactory.java
@@ -14,31 +14,42 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
package org.apache.bifromq.dist.worker.cache;
-import org.apache.bifromq.basekv.store.api.IKVCloseableReader;
import io.micrometer.core.instrument.Metrics;
import io.micrometer.core.instrument.Timer;
import java.time.Duration;
import java.util.concurrent.Executor;
import java.util.function.Supplier;
+import org.apache.bifromq.basekv.store.api.IKVCloseableReader;
+import org.apache.bifromq.plugin.eventcollector.IEventCollector;
+import org.apache.bifromq.plugin.settingprovider.ISettingProvider;
class TenantRouteCacheFactory implements ITenantRouteCacheFactory {
+ private final ISettingProvider settingProvider;
+ private final IEventCollector eventCollector;
private final Executor matchExecutor;
private final ThreadLocalKVReader threadLocalReader;
private final Timer internalMatchTimer;
private final Duration expiry;
+ private final Duration fanoutCheckInterval;
public TenantRouteCacheFactory(Supplier<IKVCloseableReader> readerSupplier,
+ ISettingProvider settingProvider,
+ IEventCollector eventCollector,
Duration expiry,
+ Duration fanoutCheckInterval,
Executor matchExecutor,
String... tags) {
+ this.settingProvider = settingProvider;
+ this.eventCollector = eventCollector;
this.matchExecutor = matchExecutor;
this.threadLocalReader = new ThreadLocalKVReader(readerSupplier);
this.expiry = expiry;
+ this.fanoutCheckInterval = fanoutCheckInterval;
internalMatchTimer = Timer.builder("dist.match.internal")
.tags(tags)
.register(Metrics.globalRegistry);
@@ -53,7 +64,8 @@ class TenantRouteCacheFactory implements ITenantRouteCacheFactory {
@Override
public ITenantRouteCache create(String tenantId) {
return new TenantRouteCache(tenantId,
- new TenantRouteMatcher(tenantId, threadLocalReader, internalMatchTimer), expiry, matchExecutor);
+ new TenantRouteMatcher(tenantId, threadLocalReader, eventCollector, internalMatchTimer),
+ settingProvider, expiry, fanoutCheckInterval, matchExecutor);
}
@Override
diff --git a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/TenantRouteMatcher.java b/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/TenantRouteMatcher.java
index 1b492917..32a18def 100644
--- a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/TenantRouteMatcher.java
+++ b/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/TenantRouteMatcher.java
@@ -14,7 +14,7 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
package org.apache.bifromq.dist.worker.cache;
@@ -27,14 +27,8 @@ import static org.apache.bifromq.basekv.utils.BoundaryUtil.upperBound;
import static org.apache.bifromq.dist.worker.schema.KVSchemaUtil.buildMatchRoute;
import static org.apache.bifromq.dist.worker.schema.KVSchemaUtil.tenantBeginKey;
import static org.apache.bifromq.dist.worker.schema.KVSchemaUtil.tenantRouteStartKey;
+import static org.apache.bifromq.plugin.eventcollector.ThreadLocalEventPool.getLocal;
-import org.apache.bifromq.basekv.proto.Boundary;
-import org.apache.bifromq.basekv.store.api.IKVIterator;
-import org.apache.bifromq.basekv.store.api.IKVReader;
-import org.apache.bifromq.dist.trie.TopicFilterIterator;
-import org.apache.bifromq.dist.trie.TopicTrieNode;
-import org.apache.bifromq.dist.worker.schema.Matching;
-import org.apache.bifromq.util.TopicUtil;
import com.google.protobuf.ByteString;
import io.micrometer.core.instrument.Timer;
import java.util.HashMap;
@@ -42,30 +36,50 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
+import org.apache.bifromq.basekv.proto.Boundary;
+import org.apache.bifromq.basekv.store.api.IKVIterator;
+import org.apache.bifromq.basekv.store.api.IKVReader;
+import org.apache.bifromq.dist.trie.TopicFilterIterator;
+import org.apache.bifromq.dist.trie.TopicTrieNode;
+import org.apache.bifromq.dist.worker.schema.Matching;
+import org.apache.bifromq.dist.worker.schema.NormalMatching;
+import org.apache.bifromq.plugin.eventcollector.IEventCollector;
+import org.apache.bifromq.plugin.eventcollector.distservice.GroupFanoutThrottled;
+import org.apache.bifromq.plugin.eventcollector.distservice.PersistentFanoutThrottled;
+import org.apache.bifromq.util.TopicUtil;
class TenantRouteMatcher implements ITenantRouteMatcher {
private final String tenantId;
private final Timer timer;
private final Supplier<IKVReader> kvReaderSupplier;
+ private final IEventCollector eventCollector;
- public TenantRouteMatcher(String tenantId, Supplier<IKVReader> kvReaderSupplier, Timer timer) {
+ public TenantRouteMatcher(String tenantId,
+ Supplier<IKVReader> kvReaderSupplier,
+ IEventCollector eventCollector,
+ Timer timer) {
this.tenantId = tenantId;
this.timer = timer;
this.kvReaderSupplier = kvReaderSupplier;
+ this.eventCollector = eventCollector;
}
@Override
- public Map<String, Set<Matching>> matchAll(Set<String> topics) {
+ public Map<String, IMatchedRoutes> matchAll(Set<String> topics,
+ int maxPersistentFanoutCount,
+ int maxGroupFanoutCount) {
final Timer.Sample sample = Timer.start();
- Map<String, Set<Matching>> matchedRoutes = new HashMap<>();
+ Map<String, IMatchedRoutes> matchedRoutes = new HashMap<>();
+ Set<String> pFanoutTrimedTopic = new HashSet<>();
+ Set<String> gFanoutTrimedTopic = new HashSet<>();
TopicTrieNode.Builder<String> topicTrieBuilder = TopicTrieNode.builder(false);
topics.forEach(topic -> {
topicTrieBuilder.addTopic(TopicUtil.parse(topic, false), topic);
- matchedRoutes.put(topic, new HashSet<>());
+ matchedRoutes.put(topic, new MatchResult(maxPersistentFanoutCount, maxGroupFanoutCount));
});
- int probe = 0;
IKVReader rangeReader = kvReaderSupplier.get();
rangeReader.refresh();
@@ -81,6 +95,7 @@ class TenantRouteMatcher implements ITenantRouteMatcher {
IKVIterator itr = rangeReader.iterator();
// track seek
itr.seek(tenantBoundary.getStartKey());
+ int probe = 0;
while (itr.isValid() && compare(itr.key(), tenantBoundary.getEndKey()) < 0) {
// track itr.key()
Matching matching = buildMatchRoute(itr.key(), itr.value());
@@ -95,7 +110,34 @@ class TenantRouteMatcher implements ITenantRouteMatcher {
Set<String> backingTopics = new HashSet<>();
for (Set<String> topicSet : expansionSetItr.value().values()) {
for (String topic : topicSet) {
- matchedRoutes.computeIfAbsent(topic, k -> new HashSet<>()).add(matching);
+ MatchResult matchResult = (MatchResult) matchedRoutes.computeIfAbsent(topic,
+ k -> new MatchResult(maxPersistentFanoutCount, maxGroupFanoutCount));
+ switch (matching.type()) {
+ case Normal -> {
+ NormalMatching normalRoute = (NormalMatching) matching;
+ if (normalRoute.subBrokerId() == 1) {
+ if (matchResult.persistentFanoutCount.get() < maxPersistentFanoutCount) {
+ matchResult.persistentFanoutCount.incrementAndGet();
+ matchResult.routes.add(matching);
+ } else {
+ pFanoutTrimedTopic.add(topic);
+ }
+ } else {
+ matchResult.routes.add(matching);
+ }
+ }
+ case Group -> {
+ if (matchResult.groupFanoutCount.get() < maxGroupFanoutCount) {
+ matchResult.groupFanoutCount.incrementAndGet();
+ matchResult.routes.add(matching);
+ } else {
+ gFanoutTrimedTopic.add(topic);
+ }
+ }
+ default -> {
+ // never happen
+ }
+ }
backingTopics.add(topic);
}
}
@@ -119,11 +161,86 @@ class TenantRouteMatcher implements ITenantRouteMatcher {
} else {
itr.next();
for (String topic : matchedTopics) {
- matchedRoutes.computeIfAbsent(topic, k -> new HashSet<>()).add(matching);
+ MatchResult matchResult = (MatchResult) matchedRoutes.computeIfAbsent(topic,
+ k -> new MatchResult(maxPersistentFanoutCount, maxGroupFanoutCount));
+ switch (matching.type()) {
+ case Normal -> {
+ NormalMatching normalRoute = (NormalMatching) matching;
+ if (normalRoute.subBrokerId() == 1) {
+ if (matchResult.persistentFanoutCount.get() < maxPersistentFanoutCount) {
+ matchResult.persistentFanoutCount.incrementAndGet();
+ matchResult.routes.add(matching);
+ } else {
+ pFanoutTrimedTopic.add(topic);
+ }
+ } else {
+ matchResult.routes.add(matching);
+ }
+ }
+ case Group -> {
+ if (matchResult.groupFanoutCount.get() < maxGroupFanoutCount) {
+ matchResult.groupFanoutCount.incrementAndGet();
+ matchResult.routes.add(matching);
+ } else {
+ gFanoutTrimedTopic.add(topic);
+ }
+ }
+ default -> {
+ // never happen
+ }
+ }
}
}
}
sample.stop(timer);
+ pFanoutTrimedTopic.forEach(topic -> eventCollector.report(getLocal(PersistentFanoutThrottled.class)
+ .tenantId(tenantId)
+ .topic(topic)
+ .maxCount(maxPersistentFanoutCount)
+ ));
+ gFanoutTrimedTopic.forEach(topic -> eventCollector.report(getLocal(GroupFanoutThrottled.class)
+ .tenantId(tenantId)
+ .topic(topic)
+ .maxCount(maxGroupFanoutCount)
+ ));
return matchedRoutes;
}
+
+ private static class MatchResult implements IMatchedRoutes {
+ private final Set<Matching> routes = new HashSet<>();
+ private final int maxPersistentFanoutCount;
+ private final int maxGroupFanoutCount;
+ private final AtomicInteger persistentFanoutCount = new AtomicInteger(0);
+ private final AtomicInteger groupFanoutCount = new AtomicInteger(0);
+
+ private MatchResult(int maxPersistentFanoutCount, int maxGroupFanoutCount) {
+ this.maxPersistentFanoutCount = maxPersistentFanoutCount;
+ this.maxGroupFanoutCount = maxGroupFanoutCount;
+ }
+
+ @Override
+ public int maxPersistentFanout() {
+ return maxPersistentFanoutCount;
+ }
+
+ @Override
+ public int maxGroupFanout() {
+ return maxGroupFanoutCount;
+ }
+
+ @Override
+ public int persistentFanout() {
+ return persistentFanoutCount.get();
+ }
+
+ @Override
+ public int groupFanout() {
+ return groupFanoutCount.get();
+ }
+
+ @Override
+ public Set<Matching> routes() {
+ return routes;
+ }
+ }
}
diff --git a/bifromq-dist/bifromq-dist-worker/src/test/java/org/apache/bifromq/dist/worker/DistQoS0Test.java b/bifromq-dist/bifromq-dist-worker/src/test/java/org/apache/bifromq/dist/worker/DistQoS0Test.java
index 4740133c..0f7180cc 100644
--- a/bifromq-dist/bifromq-dist-worker/src/test/java/org/apache/bifromq/dist/worker/DistQoS0Test.java
+++ b/bifromq-dist/bifromq-dist-worker/src/test/java/org/apache/bifromq/dist/worker/DistQoS0Test.java
@@ -14,13 +14,13 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
package org.apache.bifromq.dist.worker;
-import static org.apache.bifromq.type.QoS.AT_MOST_ONCE;
import static com.google.protobuf.ByteString.copyFromUtf8;
+import static org.apache.bifromq.type.QoS.AT_MOST_ONCE;
import static org.awaitility.Awaitility.await;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.after;
@@ -32,6 +32,12 @@ import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertTrue;
+import com.google.common.collect.Sets;
+import com.google.protobuf.ByteString;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import lombok.extern.slf4j.Slf4j;
import org.apache.bifromq.dist.rpc.proto.BatchDistReply;
import org.apache.bifromq.plugin.subbroker.DeliveryPack;
import org.apache.bifromq.plugin.subbroker.DeliveryPackage;
@@ -40,12 +46,6 @@ import org.apache.bifromq.plugin.subbroker.DeliveryResult;
import org.apache.bifromq.type.MatchInfo;
import org.apache.bifromq.type.Message;
import org.apache.bifromq.type.TopicMessagePack;
-import com.google.common.collect.Sets;
-import com.google.protobuf.ByteString;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-import lombok.extern.slf4j.Slf4j;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import org.testng.annotations.AfterMethod;
@@ -150,6 +150,7 @@ public class DistQoS0Test extends DistWorkerTest {
@Test(groups = "integration")
public void testDistCase3() {
+ // inline test
when(mqttBroker.open("batch1")).thenReturn(writer1);
match(tenantA, "/a/b/c", MqttBroker, "inbox1", "batch1");
@@ -162,6 +163,7 @@ public class DistQoS0Test extends DistWorkerTest {
assertEquals(reply.getResultMap().get(tenantA).getFanoutMap().get("/a/b/c").intValue(), 2);
ArgumentCaptor<DeliveryRequest> list1 = ArgumentCaptor.forClass(DeliveryRequest.class);
+ // the call is inlined
verify(writer1, timeout(1).times(1)).deliver(list1.capture());
log.info("Case3: verify writer1, list size is {}", list1.getAllValues().size());
int msgCount = 0;
diff --git a/bifromq-dist/bifromq-dist-worker/src/test/java/org/apache/bifromq/dist/worker/DistWorkerTest.java b/bifromq-dist/bifromq-dist-worker/src/test/java/org/apache/bifromq/dist/worker/DistWorkerTest.java
index 0b0a7923..106d2900 100644
--- a/bifromq-dist/bifromq-dist-worker/src/test/java/org/apache/bifromq/dist/worker/DistWorkerTest.java
+++ b/bifromq-dist/bifromq-dist-worker/src/test/java/org/apache/bifromq/dist/worker/DistWorkerTest.java
@@ -14,7 +14,7 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
package org.apache.bifromq.dist.worker;
@@ -35,9 +35,30 @@ import static org.awaitility.Awaitility.await;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.reset;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertTrue;
+import com.google.protobuf.ByteString;
+import io.micrometer.core.instrument.Metrics;
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
+import java.io.File;
+import java.io.IOException;
+import java.lang.reflect.Method;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.time.Duration;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadLocalRandom;
+import lombok.extern.slf4j.Slf4j;
import org.apache.bifromq.basecluster.AgentHostOptions;
import org.apache.bifromq.basecluster.IAgentHost;
import org.apache.bifromq.basecrdt.service.CRDTServiceOptions;
@@ -76,6 +97,9 @@ import org.apache.bifromq.dist.rpc.proto.DistServiceRWCoProcInput;
import org.apache.bifromq.dist.rpc.proto.MatchRoute;
import org.apache.bifromq.dist.rpc.proto.TenantOption;
import org.apache.bifromq.plugin.eventcollector.IEventCollector;
+import org.apache.bifromq.plugin.resourcethrottler.IResourceThrottler;
+import org.apache.bifromq.plugin.settingprovider.ISettingProvider;
+import org.apache.bifromq.plugin.settingprovider.Setting;
import org.apache.bifromq.plugin.subbroker.DeliveryPack;
import org.apache.bifromq.plugin.subbroker.DeliveryPackage;
import org.apache.bifromq.plugin.subbroker.DeliveryReply;
@@ -91,27 +115,6 @@ import org.apache.bifromq.type.QoS;
import org.apache.bifromq.type.RouteMatcher;
import org.apache.bifromq.type.TopicMessagePack;
import org.apache.bifromq.util.TopicUtil;
-import org.apache.bifromq.plugin.resourcethrottler.IResourceThrottler;
-import com.google.protobuf.ByteString;
-import io.micrometer.core.instrument.Metrics;
-import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
-import java.io.File;
-import java.io.IOException;
-import java.lang.reflect.Method;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.time.Duration;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.UUID;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.ThreadLocalRandom;
-import lombok.extern.slf4j.Slf4j;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
import org.mockito.stubbing.Answer;
@@ -134,6 +137,8 @@ public abstract class DistWorkerTest {
@Mock
protected IResourceThrottler resourceThrottler;
@Mock
+ protected ISettingProvider settingProvider;
+ @Mock
protected IDistClient distClient;
@Mock
protected ISubBrokerManager receiverManager;
@@ -174,6 +179,11 @@ public abstract class DistWorkerTest {
lenient().when(receiverManager.get(MqttBroker)).thenReturn(mqttBroker);
lenient().when(receiverManager.get(InboxService)).thenReturn(inboxBroker);
lenient().when(resourceThrottler.hasResource(anyString(), any())).thenReturn(true);
+ lenient().when(settingProvider.provide(any(), anyString())).thenAnswer(invocation -> {
+ Setting setting = invocation.getArgument(0);
+ String tenantId = invocation.getArgument(1);
+ return setting.current(tenantId);
+ });
bgTaskExecutor = new ScheduledThreadPoolExecutor(1,
EnvProvider.INSTANCE.newThreadFactory("bg-task-executor"));
@@ -203,7 +213,6 @@ public abstract class DistWorkerTest {
((RocksDBWALableKVEngineConfigurator) options.getWalEngineConfigurator())
.dbRootDir(Paths.get(dbRootDir.toString(), DB_WAL_NAME, uuid).toString());
-
storeClient = IBaseKVStoreClient
.newBuilder()
.eventLoopGroup(NettyEnv.createEventLoopGroup("store-client"))
@@ -224,6 +233,8 @@ public abstract class DistWorkerTest {
.bgTaskExecutor(bgTaskExecutor)
.storeOptions(options)
.subBrokerManager(receiverManager)
+ .settingProvider(settingProvider)
+ .inlineFanoutThreshold(1)
.build();
rpcServer = rpcServerBuilder.build();
rpcServer.start();
@@ -265,6 +276,7 @@ public abstract class DistWorkerTest {
public void printCaseFinish(Method method) {
log.info("Test case[{}.{}] finished, doing teardown",
method.getDeclaringClass().getName(), method.getName());
+ reset(eventCollector);
}
protected BatchMatchReply.TenantBatch.Code match(String tenantId,
diff --git a/bifromq-dist/bifromq-dist-worker/src/test/java/org/apache/bifromq/dist/worker/FanoutThrottledTest.java b/bifromq-dist/bifromq-dist-worker/src/test/java/org/apache/bifromq/dist/worker/FanoutThrottledTest.java
new file mode 100644
index 00000000..eb1f21e9
--- /dev/null
+++ b/bifromq-dist/bifromq-dist-worker/src/test/java/org/apache/bifromq/dist/worker/FanoutThrottledTest.java
@@ -0,0 +1,263 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.bifromq.dist.worker;
+
+import static com.google.protobuf.ByteString.copyFromUtf8;
+import static org.apache.bifromq.type.QoS.AT_MOST_ONCE;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.argThat;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.atLeast;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+
+import java.util.concurrent.CompletableFuture;
+import org.apache.bifromq.dist.rpc.proto.BatchDistReply;
+import org.apache.bifromq.plugin.eventcollector.EventType;
+import org.apache.bifromq.plugin.eventcollector.OutOfTenantResource;
+import org.apache.bifromq.plugin.eventcollector.distservice.GroupFanoutThrottled;
+import org.apache.bifromq.plugin.eventcollector.distservice.PersistentFanoutBytesThrottled;
+import org.apache.bifromq.plugin.eventcollector.distservice.PersistentFanoutThrottled;
+import org.apache.bifromq.plugin.resourcethrottler.TenantResourceType;
+import org.apache.bifromq.plugin.settingprovider.Setting;
+import org.apache.bifromq.plugin.subbroker.DeliveryReply;
+import org.testng.annotations.Test;
+
+public class FanoutThrottledTest extends DistWorkerTest {
+ @Test(groups = "integration")
+ public void onePassPersistentFanoutLimit() {
+ when(inboxBroker.open("batch1")).thenReturn(writer1);
+ when(inboxBroker.open("batch2")).thenReturn(writer2);
+ when(inboxBroker.open("batch3")).thenReturn(writer3);
+ when(writer1.deliver(any())).thenReturn(CompletableFuture.completedFuture(DeliveryReply.getDefaultInstance()));
+ when(writer2.deliver(any())).thenReturn(CompletableFuture.completedFuture(DeliveryReply.getDefaultInstance()));
+ when(writer3.deliver(any())).thenReturn(CompletableFuture.completedFuture(DeliveryReply.getDefaultInstance()));
+
+ // match 3 persistent fanout subscriptions
+ match(tenantA, "/fanout/topic", InboxService, "inbox1", "batch1");
+ match(tenantA, "/fanout/topic", InboxService, "inbox2", "batch2");
+ match(tenantA, "/fanout/topic", InboxService, "inbox3", "batch3");
+
+ when(settingProvider.provide(eq(Setting.MaxPersistentFanout), any())).thenReturn(2);
+
+ // trigger dist
+ BatchDistReply reply = dist(tenantA, AT_MOST_ONCE, "/fanout/topic", copyFromUtf8("throttled"), "orderKey1");
+
+ // verify reply shows fanout = 2
+ assertEquals(reply.getResultMap().get(tenantA).getFanoutMap().get("/fanout/topic").intValue(), 2);
+ // cleanup
+ unmatch(tenantA, "/fanout/topic", MqttBroker, "inbox1", "batch1");
+ unmatch(tenantA, "/fanout/topic", MqttBroker, "inbox2", "batch2");
+ unmatch(tenantA, "/fanout/topic", MqttBroker, "inbox3", "batch3");
+ }
+
+ @Test(groups = "integration")
+ public void deliverPersistentFanoutThrottled() {
+ when(inboxBroker.open("batch1")).thenReturn(writer1);
+ when(inboxBroker.open("batch2")).thenReturn(writer2);
+ when(inboxBroker.open("batch3")).thenReturn(writer3);
+ when(writer1.deliver(any())).thenReturn(CompletableFuture.completedFuture(DeliveryReply.getDefaultInstance()));
+ when(writer2.deliver(any())).thenReturn(CompletableFuture.completedFuture(DeliveryReply.getDefaultInstance()));
+ when(writer3.deliver(any())).thenReturn(CompletableFuture.completedFuture(DeliveryReply.getDefaultInstance()));
+
+ // match 3 persistent fanout subscriptions
+ match(tenantA, "/fanout/topic2", InboxService, "inbox1", "batch1");
+ match(tenantA, "/fanout/topic2", InboxService, "inbox2", "batch2");
+ match(tenantA, "/fanout/topic2", InboxService, "inbox3", "batch3");
+ // trigger dist
+ BatchDistReply reply = dist(tenantA, AT_MOST_ONCE, "/fanout/topic2", copyFromUtf8("throttled"), "orderKey1");
+ // verify reply shows fanout = 3
+ assertEquals(reply.getResultMap().get(tenantA).getFanoutMap().get("/fanout/topic2").intValue(), 3);
+
+ // set max persistent fanout = 2
+ when(settingProvider.provide(eq(Setting.MaxPersistentFanout), any())).thenReturn(2);
+ when(settingProvider.provide(eq(Setting.MaxPersistentFanoutBytes), any())).thenReturn(Long.MAX_VALUE);
+
+ // trigger dist and cache will be used
+ dist(tenantA, AT_MOST_ONCE, "/fanout/topic2", copyFromUtf8("throttled"), "orderKey1");
+
+ verify(eventCollector, atLeast(1)).report(argThat(e -> e.type() == EventType.PERSISTENT_FANOUT_THROTTLED
+ && ((PersistentFanoutThrottled) e).tenantId().equals(tenantA)
+ && ((PersistentFanoutThrottled) e).topic().equals("/fanout/topic2")
+ && ((PersistentFanoutThrottled) e).maxCount() == 2));
+ // cleanup
+ unmatch(tenantA, "/fanout/topic2", MqttBroker, "inbox1", "batch1");
+ unmatch(tenantA, "/fanout/topic2", MqttBroker, "inbox2", "batch2");
+ unmatch(tenantA, "/fanout/topic2", MqttBroker, "inbox3", "batch3");
+ }
+
+ @Test(groups = "integration")
+ public void deliverPersistentFanoutBytesThrottled() {
+ when(inboxBroker.open("batch1")).thenReturn(writer1);
+ when(inboxBroker.open("batch2")).thenReturn(writer2);
+ when(inboxBroker.open("batch3")).thenReturn(writer3);
+ when(writer1.deliver(any())).thenReturn(CompletableFuture.completedFuture(DeliveryReply.getDefaultInstance()));
+ when(writer2.deliver(any())).thenReturn(CompletableFuture.completedFuture(DeliveryReply.getDefaultInstance()));
+ when(writer3.deliver(any())).thenReturn(CompletableFuture.completedFuture(DeliveryReply.getDefaultInstance()));
+
+ // match 3 persistent fanout subscriptions
+ match(tenantA, "/fanout/topic", InboxService, "inbox1", "batch1");
+ match(tenantA, "/fanout/topic", InboxService, "inbox2", "batch2");
+ match(tenantA, "/fanout/topic", InboxService, "inbox3", "batch3");
+ // trigger dist
+ BatchDistReply reply = dist(tenantA, AT_MOST_ONCE, "/fanout/topic", copyFromUtf8("throttled"), "orderKey1");
+ // verify reply shows fanout = 3
+ assertEquals(reply.getResultMap().get(tenantA).getFanoutMap().get("/fanout/topic").intValue(), 3);
+
+ when(settingProvider.provide(eq(Setting.MaxPersistentFanout), any())).thenReturn(Integer.MAX_VALUE);
+ when(settingProvider.provide(eq(Setting.MaxPersistentFanoutBytes), any())).thenReturn(10L);
+
+ // trigger dist and cache will be used
+ dist(tenantA, AT_MOST_ONCE, "/fanout/topic", copyFromUtf8("throttled"), "orderKey1");
+
+ verify(eventCollector).report(argThat(e -> e.type() == EventType.PERSISTENT_FANOUT_BYTES_THROTTLED
+ && ((PersistentFanoutBytesThrottled) e).tenantId().equals(tenantA)
+ && ((PersistentFanoutBytesThrottled) e).topic().equals("/fanout/topic")
+ && ((PersistentFanoutBytesThrottled) e).maxBytes() == 10));
+ // cleanup
+ unmatch(tenantA, "/fanout/topic", MqttBroker, "inbox1", "batch1");
+ unmatch(tenantA, "/fanout/topic", MqttBroker, "inbox2", "batch2");
+ unmatch(tenantA, "/fanout/topic", MqttBroker, "inbox3", "batch3");
+ }
+
+ @Test(groups = "integration")
+ public void onePassGroupFanoutLimit() {
+ when(inboxBroker.open("batch1")).thenReturn(writer1);
+ when(inboxBroker.open("batch2")).thenReturn(writer2);
+ when(inboxBroker.open("batch3")).thenReturn(writer3);
+ when(writer1.deliver(any())).thenReturn(CompletableFuture.completedFuture(DeliveryReply.getDefaultInstance()));
+ when(writer2.deliver(any())).thenReturn(CompletableFuture.completedFuture(DeliveryReply.getDefaultInstance()));
+ when(writer3.deliver(any())).thenReturn(CompletableFuture.completedFuture(DeliveryReply.getDefaultInstance()));
+
+ // match 3 group fanout subscriptions
+ match(tenantA, "$share/group1/fanout/topic", InboxService, "inbox1", "batch1");
+ match(tenantA, "$share/group2/fanout/topic", InboxService, "inbox2", "batch2");
+ match(tenantA, "$share/group3/fanout/topic", InboxService, "inbox3", "batch3");
+
+ when(settingProvider.provide(eq(Setting.MaxGroupFanout), any())).thenReturn(2);
+
+ // trigger dist
+ BatchDistReply reply = dist(tenantA, AT_MOST_ONCE, "fanout/topic", copyFromUtf8("throttled"), "orderKey1");
+
+ // verify reply shows fanout = 2
+ assertEquals(reply.getResultMap().get(tenantA).getFanoutMap().get("fanout/topic").intValue(), 2);
+ // cleanup
+ unmatch(tenantA, "$share/group1/fanout/topic", MqttBroker, "inbox1", "batch1");
+ unmatch(tenantA, "$share/group2/fanout/topic", MqttBroker, "inbox2", "batch2");
+ unmatch(tenantA, "$share/group3/fanout/topic", MqttBroker, "inbox3", "batch3");
+ }
+
+ @Test(groups = "integration")
+ public void deliverGroupFanoutThrottled() {
+ when(inboxBroker.open("batch1")).thenReturn(writer1);
+ when(inboxBroker.open("batch2")).thenReturn(writer2);
+ when(inboxBroker.open("batch3")).thenReturn(writer3);
+ when(writer1.deliver(any())).thenReturn(CompletableFuture.completedFuture(DeliveryReply.getDefaultInstance()));
+ when(writer2.deliver(any())).thenReturn(CompletableFuture.completedFuture(DeliveryReply.getDefaultInstance()));
+ when(writer3.deliver(any())).thenReturn(CompletableFuture.completedFuture(DeliveryReply.getDefaultInstance()));
+
+ // match 3 group fanout subscriptions
+ match(tenantA, "$share/group1/fanout/topic", InboxService, "inbox1", "batch1");
+ match(tenantA, "$share/group2/fanout/topic", InboxService, "inbox2", "batch2");
+ match(tenantA, "$share/group3/fanout/topic", InboxService, "inbox3", "batch3");
+ // trigger dist
+ BatchDistReply reply = dist(tenantA, AT_MOST_ONCE, "fanout/topic", copyFromUtf8("throttled"), "orderKey1");
+ // verify reply shows fanout = 3
+ assertEquals(reply.getResultMap().get(tenantA).getFanoutMap().get("fanout/topic").intValue(), 3);
+
+ // set max group fanout = 2
+ when(settingProvider.provide(eq(Setting.MaxGroupFanout), any())).thenReturn(2);
+
+ // trigger dist and cache will be used
+ dist(tenantA, AT_MOST_ONCE, "fanout/topic", copyFromUtf8("throttled"), "orderKey1");
+
+ verify(eventCollector).report(argThat(e -> e.type() == EventType.GROUP_FANOUT_THROTTLED
+ && ((GroupFanoutThrottled) e).tenantId().equals(tenantA)
+ && ((GroupFanoutThrottled) e).topic().equals("fanout/topic")
+ && ((GroupFanoutThrottled) e).maxCount() == 2));
+ // cleanup
+ unmatch(tenantA, "$share/group1/fanout/topic", MqttBroker, "inbox1", "batch1");
+ unmatch(tenantA, "$share/group2/fanout/topic", MqttBroker, "inbox2", "batch2");
+ unmatch(tenantA, "$share/group3/fanout/topic", MqttBroker, "inbox3", "batch3");
+ }
+
+ @Test(groups = "integration")
+ public void deliverNoPersistentFanoutBandwidth() {
+ when(inboxBroker.open("batch1")).thenReturn(writer1);
+ when(inboxBroker.open("batch2")).thenReturn(writer2);
+ when(inboxBroker.open("batch3")).thenReturn(writer3);
+ when(writer1.deliver(any())).thenReturn(CompletableFuture.completedFuture(DeliveryReply.getDefaultInstance()));
+ when(writer2.deliver(any())).thenReturn(CompletableFuture.completedFuture(DeliveryReply.getDefaultInstance()));
+ when(writer3.deliver(any())).thenReturn(CompletableFuture.completedFuture(DeliveryReply.getDefaultInstance()));
+
+ // match 3 persistent fanout subscriptions
+ match(tenantA, "/fanout/topic2", InboxService, "inbox1", "batch1");
+ match(tenantA, "/fanout/topic2", InboxService, "inbox2", "batch2");
+ match(tenantA, "/fanout/topic2", InboxService, "inbox3", "batch3");
+
+ // set no persistent fanout bandwidth
+ when(resourceThrottler.hasResource(eq(tenantA), eq(TenantResourceType.TotalPersistentFanOutBytesPerSeconds)))
+ .thenReturn(false);
+ // trigger dist
+ BatchDistReply reply = dist(tenantA, AT_MOST_ONCE, "/fanout/topic2", copyFromUtf8("throttled"), "orderKey1");
+ // verify reply shows fanout = 3
+ assertEquals(reply.getResultMap().get(tenantA).getFanoutMap().get("/fanout/topic2").intValue(), 3);
+
+ verify(eventCollector, atLeast(1))
+ .report(argThat(e -> e.type() == EventType.OUT_OF_TENANT_RESOURCE
+ && ((OutOfTenantResource) e).reason().equals("TotalPersistentFanOutBytesPerSeconds")));
+ // cleanup
+ unmatch(tenantA, "/fanout/topic2", MqttBroker, "inbox1", "batch1");
+ unmatch(tenantA, "/fanout/topic2", MqttBroker, "inbox2", "batch2");
+ unmatch(tenantA, "/fanout/topic2", MqttBroker, "inbox3", "batch3");
+ }
+
+ @Test(groups = "integration")
+ public void deliverNoTransientFanoutBandwidth() {
+ when(mqttBroker.open("batch1")).thenReturn(writer1);
+ when(mqttBroker.open("batch2")).thenReturn(writer2);
+ when(mqttBroker.open("batch3")).thenReturn(writer3);
+ when(writer1.deliver(any())).thenReturn(CompletableFuture.completedFuture(DeliveryReply.getDefaultInstance()));
+ when(writer2.deliver(any())).thenReturn(CompletableFuture.completedFuture(DeliveryReply.getDefaultInstance()));
+ when(writer3.deliver(any())).thenReturn(CompletableFuture.completedFuture(DeliveryReply.getDefaultInstance()));
+
+ // match 3 transient fanout subscriptions
+ match(tenantA, "/fanout/topic2", MqttBroker, "inbox1", "batch1");
+ match(tenantA, "/fanout/topic2", MqttBroker, "inbox2", "batch2");
+ match(tenantA, "/fanout/topic2", MqttBroker, "inbox3", "batch3");
+
+ // set no persistent fanout bandwidth
+ when(resourceThrottler.hasResource(eq(tenantA), eq(TenantResourceType.TotalTransientFanOutBytesPerSeconds)))
+ .thenReturn(false);
+ // trigger dist
+ BatchDistReply reply = dist(tenantA, AT_MOST_ONCE, "/fanout/topic2", copyFromUtf8("throttled"), "orderKey1");
+ // verify reply shows fanout = 3
+ assertEquals(reply.getResultMap().get(tenantA).getFanoutMap().get("/fanout/topic2").intValue(), 3);
+
+ verify(eventCollector, atLeast(1))
+ .report(argThat(e -> e.type() == EventType.OUT_OF_TENANT_RESOURCE
+ && ((OutOfTenantResource) e).reason().equals("TotalTransientFanOutBytesPerSeconds")));
+ // cleanup
+ unmatch(tenantA, "/fanout/topic2", MqttBroker, "inbox1", "batch1");
+ unmatch(tenantA, "/fanout/topic2", MqttBroker, "inbox2", "batch2");
+ unmatch(tenantA, "/fanout/topic2", MqttBroker, "inbox3", "batch3");
+ }
+}
diff --git a/bifromq-dist/bifromq-dist-worker/src/test/java/org/apache/bifromq/dist/worker/cache/TenantRouteCacheTest.java b/bifromq-dist/bifromq-dist-worker/src/test/java/org/apache/bifromq/dist/worker/cache/TenantRouteCacheTest.java
index 1d10c201..0efa00f4 100644
--- a/bifromq-dist/bifromq-dist-worker/src/test/java/org/apache/bifromq/dist/worker/cache/TenantRouteCacheTest.java
+++ b/bifromq-dist/bifromq-dist-worker/src/test/java/org/apache/bifromq/dist/worker/cache/TenantRouteCacheTest.java
@@ -14,33 +14,37 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
package org.apache.bifromq.dist.worker.cache;
-import static org.apache.bifromq.basekv.utils.BoundaryUtil.FULL_BOUNDARY;
import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
+import static org.apache.bifromq.basekv.utils.BoundaryUtil.FULL_BOUNDARY;
import static org.awaitility.Awaitility.await;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
-import org.apache.bifromq.dist.worker.Comparators;
-import org.apache.bifromq.dist.worker.MeterTest;
-import org.apache.bifromq.dist.worker.schema.Matching;
-import org.apache.bifromq.metrics.TenantMetric;
-import org.apache.bifromq.type.RouteMatcher;
-import org.apache.bifromq.util.TopicUtil;
import com.github.benmanes.caffeine.cache.Ticker;
import com.google.common.collect.Sets;
import java.time.Duration;
-import java.util.Collections;
import java.util.Map;
import java.util.NavigableSet;
import java.util.Set;
+import org.apache.bifromq.dist.worker.Comparators;
+import org.apache.bifromq.dist.worker.MeterTest;
+import org.apache.bifromq.dist.worker.schema.Matching;
+import org.apache.bifromq.metrics.TenantMetric;
+import org.apache.bifromq.plugin.settingprovider.ISettingProvider;
+import org.apache.bifromq.plugin.settingprovider.Setting;
+import org.apache.bifromq.type.RouteMatcher;
+import org.apache.bifromq.util.TopicUtil;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@@ -48,19 +52,24 @@ import org.testng.annotations.Test;
public class TenantRouteCacheTest extends MeterTest {
private TenantRouteCache cache;
- private Ticker mockTicker;
+ private ManualTicker mockTicker;
private ITenantRouteMatcher mockMatcher;
+ private ISettingProvider mockSettingProvider;
private String tenantId;
private Duration expiryDuration;
+ private Duration fanoutCheckDuration;
@BeforeMethod
public void setup() {
super.setup();
tenantId = "tenant1";
- mockTicker = mock(Ticker.class);
+ mockTicker = new ManualTicker();
mockMatcher = mock(ITenantRouteMatcher.class);
+ mockSettingProvider = mock(ISettingProvider.class);
expiryDuration = Duration.ofMinutes(1);
- cache = new TenantRouteCache(tenantId, mockMatcher, expiryDuration, directExecutor());
+ fanoutCheckDuration = Duration.ofSeconds(2);
+ cache = new TenantRouteCache(tenantId, mockMatcher, mockSettingProvider, expiryDuration,
+ fanoutCheckDuration, directExecutor());
}
@AfterMethod
@@ -69,22 +78,27 @@ public class TenantRouteCacheTest extends MeterTest {
super.tearDown();
}
+ @SuppressWarnings("checkstyle:WhitespaceAround")
@Test
public void getMatch() {
String topic = "home/sensor/temperature";
- Matching normalMatching = mock(Matching.class);
- when(mockMatcher.matchAll(eq(Set.of(topic)))).thenReturn(Map.of(topic, Set.of(normalMatching)));
+ when(mockSettingProvider.provide(eq(Setting.MaxPersistentFanout), eq(tenantId))).thenReturn(10);
+ when(mockSettingProvider.provide(eq(Setting.MaxGroupFanout), eq(tenantId))).thenReturn(5);
+ when(mockMatcher.matchAll(eq(Set.of(topic)), eq(10), eq(5)))
+ .thenReturn(Map.of(topic, mockResult(1, 0, 0)));
Set<Matching> cachedMatchings = cache.getMatch(topic, FULL_BOUNDARY).join();
assertNotNull(cachedMatchings);
- assertTrue(cachedMatchings.contains(normalMatching));
+ assertEquals(cachedMatchings.size(), 1);
}
@Test
public void isCached() {
String topic = "home/sensor/temperature";
- Matching normalMatching = mock(Matching.class);
- when(mockMatcher.matchAll(eq(Set.of(topic)))).thenReturn(Map.of(topic, Set.of(normalMatching)));
+ when(mockSettingProvider.provide(eq(Setting.MaxPersistentFanout), eq(tenantId))).thenReturn(10);
+ when(mockSettingProvider.provide(eq(Setting.MaxGroupFanout), eq(tenantId))).thenReturn(5);
+ when(mockMatcher.matchAll(eq(Set.of(topic)), eq(10), eq(5)))
+ .thenReturn(Map.of(topic, mockResult(0, 0, 0)));
cache.getMatch(topic, FULL_BOUNDARY).join();
assertTrue(cache.isCached(TopicUtil.from(topic).getFilterLevelList()));
@@ -93,59 +107,233 @@ public class TenantRouteCacheTest extends MeterTest {
@Test
public void refreshToAddMatch() {
String topic = "home/sensor/temperature";
- Matching normalMatching = mock(Matching.class);
- when(normalMatching.type()).thenReturn(Matching.Type.Normal);
- when(mockMatcher.matchAll(eq(Set.of(topic)))).thenReturn(Map.of(topic, Set.of(normalMatching)));
+ when(mockSettingProvider.provide(eq(Setting.MaxPersistentFanout), eq(tenantId))).thenReturn(10);
+ when(mockSettingProvider.provide(eq(Setting.MaxGroupFanout), eq(tenantId))).thenReturn(5);
+ when(mockMatcher.matchAll(eq(Set.of(topic)), eq(10), eq(5)))
+ .thenReturn(
+ Map.of(topic, mockResult(1, 0, 0)),
+ Map.of(topic, mockResult(3, 0, 0))
+ );
Set<Matching> cachedMatchings = cache.getMatch(topic, FULL_BOUNDARY).join();
- assertTrue(cachedMatchings.contains(normalMatching));
+ assertEquals(cachedMatchings.size(), 1);
- Matching normalMatching1 = mock(Matching.class);
- when(normalMatching1.type()).thenReturn(Matching.Type.Normal);
+ NavigableSet<RouteMatcher> routeMatchers = Sets.newTreeSet(Comparators.RouteMatcherComparator);
+ routeMatchers.add(TopicUtil.from("#"));
+ cache.refresh(routeMatchers);
- Matching normalMatching2 = mock(Matching.class);
- when(normalMatching2.type()).thenReturn(Matching.Type.Normal);
+ await().until(() -> cache.getMatch(topic, FULL_BOUNDARY).join().size() == 3);
+ }
- when(mockMatcher.matchAll(eq(Set.of(topic)))).thenReturn(
- Map.of(topic, Set.of(normalMatching, normalMatching1, normalMatching2)));
+ @Test
+ public void refreshToRemoveMatch() {
+ String topic = "home/sensor/temperature";
+ when(mockSettingProvider.provide(eq(Setting.MaxPersistentFanout), eq(tenantId))).thenReturn(10);
+ when(mockSettingProvider.provide(eq(Setting.MaxGroupFanout), eq(tenantId))).thenReturn(5);
+ when(mockMatcher.matchAll(eq(Set.of(topic)), eq(10), eq(5)))
+ .thenReturn(Map.of(topic, mockResult(2, 0, 0)));
+
+ Set<Matching> cachedMatchings = cache.getMatch(topic, FULL_BOUNDARY).join();
+ assertEquals(cachedMatchings.size(), 2);
+
+ when(mockMatcher.matchAll(eq(Set.of(topic)), eq(10), eq(5)))
+ .thenReturn(Map.of(topic, mockResult(0, 0, 0)));
NavigableSet<RouteMatcher> routeMatchers = Sets.newTreeSet(Comparators.RouteMatcherComparator);
routeMatchers.add(TopicUtil.from("#"));
cache.refresh(routeMatchers);
- await().until(() -> cache.getMatch(topic, FULL_BOUNDARY).join().size() == 3);
+ await().until(() -> cache.getMatch(topic, FULL_BOUNDARY).join().isEmpty());
}
@Test
- public void refreshToRemoveMatch() {
+ public void testPersistentFanoutCheck() {
String topic = "home/sensor/temperature";
- Matching normalMatching = mock(Matching.class);
- when(normalMatching.type()).thenReturn(Matching.Type.Normal);
+ when(mockSettingProvider.provide(eq(Setting.MaxPersistentFanout), eq(tenantId))).thenReturn(10, 5, 10);
+ when(mockSettingProvider.provide(eq(Setting.MaxGroupFanout), eq(tenantId))).thenReturn(5);
+ when(mockMatcher.matchAll(eq(Set.of(topic)), eq(10), eq(5)))
+ .thenReturn(Map.of(topic, mockResult(0, 10, 0)));
- Matching normalMatching1 = mock(Matching.class);
- when(normalMatching1.type()).thenReturn(Matching.Type.Normal);
+ Set<Matching> cachedMatchings = cache.getMatch(topic, FULL_BOUNDARY).join();
+ assertEquals(cachedMatchings.size(), 10);
- when(mockMatcher.matchAll(eq(Set.of(topic)))).thenReturn(
- Map.of(topic, Sets.newHashSet(normalMatching, normalMatching1)));
+ when(mockMatcher.matchAll(eq(Set.of(topic)), eq(5), eq(5)))
+ .thenReturn(Map.of(topic, mockResult(0, 5, 0)));
+
+ NavigableSet<RouteMatcher> routeMatchers = Sets.newTreeSet(Comparators.RouteMatcherComparator);
+ routeMatchers.add(TopicUtil.from("#"));
+ // set maxPersistentFanout to 5
+ cache.refresh(routeMatchers);
+ await().until(() -> cache.getMatch(topic, FULL_BOUNDARY).join().size() == 5);
+
+ // set maxPersistentFanout to 10
+ cache.refresh(routeMatchers);
+
+ await().until(() -> cache.getMatch(topic, FULL_BOUNDARY).join().size() == 10);
+ }
+
+ @Test
+ public void testGroupFanoutCheck() {
+ String topic = "home/sensor/temperature";
+ when(mockSettingProvider.provide(eq(Setting.MaxPersistentFanout), eq(tenantId))).thenReturn(10);
+ when(mockSettingProvider.provide(eq(Setting.MaxGroupFanout), eq(tenantId))).thenReturn(5, 3, 5);
+ when(mockMatcher.matchAll(eq(Set.of(topic)), eq(10), eq(5)))
+ .thenReturn(Map.of(topic, mockResult(0, 0, 5)));
Set<Matching> cachedMatchings = cache.getMatch(topic, FULL_BOUNDARY).join();
- assertTrue(cachedMatchings.contains(normalMatching));
- assertTrue(cachedMatchings.contains(normalMatching1));
+ assertEquals(cachedMatchings.size(), 5);
- when(mockMatcher.matchAll(eq(Set.of(topic)))).thenReturn(Map.of(topic, Collections.emptySet()));
+ when(mockMatcher.matchAll(eq(Set.of(topic)), eq(10), eq(3)))
+ .thenReturn(Map.of(topic, mockResult(0, 0, 3)));
NavigableSet<RouteMatcher> routeMatchers = Sets.newTreeSet(Comparators.RouteMatcherComparator);
routeMatchers.add(TopicUtil.from("#"));
+ // set maxGroupFanout to 3
cache.refresh(routeMatchers);
+ await().until(() -> cache.getMatch(topic, FULL_BOUNDARY).join().size() == 3);
- await().until(() -> cache.getMatch(topic, FULL_BOUNDARY).join().isEmpty());
+ // set maxGroupFanout to 5
+ cache.refresh(routeMatchers);
+ await().until(() -> cache.getMatch(topic, FULL_BOUNDARY).join().size() == 5);
+ }
+
+ @Test
+ public void reloadOnPersistentFanoutLimitDecrease() {
+ String topic = "home/sensor/temperature";
+ when(mockSettingProvider.provide(eq(Setting.MaxPersistentFanout), eq(tenantId))).thenReturn(10, 5);
+ when(mockSettingProvider.provide(eq(Setting.MaxGroupFanout), eq(tenantId))).thenReturn(5);
+
+ when(mockMatcher.matchAll(eq(Set.of(topic)), eq(10), eq(5)))
+ .thenReturn(Map.of(topic, mockResult(0, 10, 0)));
+
+ assertEquals(cache.getMatch(topic, FULL_BOUNDARY).join().size(), 10);
+
+ when(mockMatcher.matchAll(eq(Set.of(topic)), eq(5), eq(5)))
+ .thenReturn(Map.of(topic, mockResult(0, 5, 0)));
+
+ mockTicker.advance(Duration.ofMillis(10));
+ await().until(() -> cache.getMatch(topic, FULL_BOUNDARY).join().size() == 5);
+ }
+
+ @Test
+ public void reloadOnPersistentFanoutLimitIncreaseWhenClamped() {
+ String topic = "home/sensor/temperature";
+ when(mockSettingProvider.provide(eq(Setting.MaxPersistentFanout), eq(tenantId))).thenReturn(5, 10);
+ when(mockSettingProvider.provide(eq(Setting.MaxGroupFanout), eq(tenantId))).thenReturn(5);
+
+ when(mockMatcher.matchAll(eq(Set.of(topic)), eq(5), eq(5)))
+ .thenReturn(Map.of(topic, mockResult(0, 5, 0)));
+
+ assertEquals(cache.getMatch(topic, FULL_BOUNDARY).join().size(), 5);
+
+ when(mockMatcher.matchAll(eq(Set.of(topic)), eq(10), eq(5)))
+ .thenReturn(Map.of(topic, mockResult(0, 10, 0)));
+
+ mockTicker.advance(Duration.ofMillis(10));
+ await().until(() -> cache.getMatch(topic, FULL_BOUNDARY).join().size() == 10);
+ }
+
+ @Test
+ public void reloadDoesNotRematchWhenIncreaseWithoutClamp() {
+ String topic = "home/sensor/temperature";
+ when(mockSettingProvider.provide(eq(Setting.MaxPersistentFanout), eq(tenantId))).thenReturn(10, 20);
+ when(mockSettingProvider.provide(eq(Setting.MaxGroupFanout), eq(tenantId))).thenReturn(5);
+ when(mockMatcher.matchAll(eq(Set.of(topic)), eq(10), eq(5)))
+ .thenReturn(Map.of(topic, mockResult(0, 3, 0)));
+
+ Set<Matching> first = cache.getMatch(topic, FULL_BOUNDARY).join();
+ assertEquals(first.size(), 3);
+
+ mockTicker.advance(Duration.ofMillis(10));
+
+ Set<Matching> second = cache.getMatch(topic, FULL_BOUNDARY).join();
+ assertEquals(second.size(), 3);
+
+ verify(mockMatcher, times(1)).matchAll(eq(Set.of(topic)), eq(10), eq(5));
}
@Test
- void testDestroy() {
- TenantRouteCache cache = new TenantRouteCache(tenantId, mockMatcher, expiryDuration, directExecutor());
+ public void testDestroy() {
+ TenantRouteCache cache = new TenantRouteCache(tenantId, mockMatcher, mockSettingProvider,
+ expiryDuration, fanoutCheckDuration, directExecutor());
assertGauge(tenantId, TenantMetric.MqttRouteCacheSize);
cache.destroy();
assertNoGauge(tenantId, TenantMetric.MqttRouteCacheSize);
}
+
+ private IMatchedRoutes mockResult(int transientFanout, int persistentFanout, int groupFanout) {
+ Set<Matching> matchings = Sets.newHashSet();
+ for (int i = 0; i < transientFanout; i++) {
+ Matching transientMatching = new Matching(tenantId, RouteMatcher.newBuilder()
+ .setMqttTopicFilter("transient/topic/" + i)
+ .build()) {
+ @Override
+ public Type type() {
+ return Matching.Type.Normal;
+ }
+ };
+ matchings.add(transientMatching);
+ }
+ for (int i = 0; i < persistentFanout; i++) {
+ Matching transientMatching = new Matching(tenantId, RouteMatcher.newBuilder()
+ .setMqttTopicFilter("transient/topic/" + i)
+ .build()) {
+ @Override
+ public Type type() {
+ return Matching.Type.Normal;
+ }
+ };
+ matchings.add(transientMatching);
+ }
+ for (int i = 0; i < groupFanout; i++) {
+ Matching transientMatching = new Matching(tenantId, RouteMatcher.newBuilder()
+ .setMqttTopicFilter("transient/topic/" + i)
+ .build()) {
+ @Override
+ public Type type() {
+ return Type.Group;
+ }
+ };
+ matchings.add(transientMatching);
+ }
+ return new IMatchedRoutes() {
+ @Override
+ public int maxPersistentFanout() {
+ return persistentFanout;
+ }
+
+ @Override
+ public int maxGroupFanout() {
+ return groupFanout;
+ }
+
+ @Override
+ public int persistentFanout() {
+ return persistentFanout;
+ }
+
+ @Override
+ public int groupFanout() {
+ return groupFanout;
+ }
+
+ @Override
+ public Set<Matching> routes() {
+ return matchings;
+ }
+ };
+ }
+
+ private static final class ManualTicker implements Ticker {
+ private long nanos = 0L;
+
+ @Override
+ public long read() {
+ return nanos;
+ }
+
+ void advance(Duration d) {
+ nanos += d.toNanos();
+ }
+ }
}
\ No newline at end of file
diff --git a/bifromq-mqtt/bifromq-mqtt-server/src/test/java/org/apache/bifromq/mqtt/integration/MQTTTest.java b/bifromq-mqtt/bifromq-mqtt-server/src/test/java/org/apache/bifromq/mqtt/integration/MQTTTest.java
index b9c551d6..855fe427 100644
--- a/bifromq-mqtt/bifromq-mqtt-server/src/test/java/org/apache/bifromq/mqtt/integration/MQTTTest.java
+++ b/bifromq-mqtt/bifromq-mqtt-server/src/test/java/org/apache/bifromq/mqtt/integration/MQTTTest.java
@@ -14,7 +14,7 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
package org.apache.bifromq.mqtt.integration;
@@ -24,22 +24,6 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.lenient;
-import org.apache.bifromq.baserpc.client.IRPCClient;
-import org.apache.bifromq.baserpc.server.IRPCServer;
-import org.apache.bifromq.baserpc.server.RPCServerBuilder;
-import org.apache.bifromq.baserpc.trafficgovernor.IRPCServiceTrafficService;
-import org.apache.bifromq.plugin.authprovider.IAuthProvider;
-import org.apache.bifromq.plugin.clientbalancer.IClientBalancer;
-import org.apache.bifromq.plugin.eventcollector.Event;
-import org.apache.bifromq.plugin.eventcollector.IEventCollector;
-import org.apache.bifromq.plugin.settingprovider.ISettingProvider;
-import org.apache.bifromq.plugin.settingprovider.Setting;
-import org.apache.bifromq.plugin.subbroker.ISubBrokerManager;
-import org.apache.bifromq.plugin.subbroker.SubBrokerManager;
-import org.apache.bifromq.retain.client.IRetainClient;
-import org.apache.bifromq.retain.server.IRetainServer;
-import org.apache.bifromq.retain.store.IRetainStore;
-import org.apache.bifromq.plugin.resourcethrottler.IResourceThrottler;
import com.google.common.collect.Sets;
import io.reactivex.rxjava3.core.Observable;
import java.lang.reflect.Method;
@@ -58,6 +42,10 @@ import org.apache.bifromq.basekv.localengine.memory.InMemKVEngineConfigurator;
import org.apache.bifromq.basekv.metaservice.IBaseKVMetaService;
import org.apache.bifromq.basekv.store.option.KVRangeStoreOptions;
import org.apache.bifromq.basekv.utils.BoundaryUtil;
+import org.apache.bifromq.baserpc.client.IRPCClient;
+import org.apache.bifromq.baserpc.server.IRPCServer;
+import org.apache.bifromq.baserpc.server.RPCServerBuilder;
+import org.apache.bifromq.baserpc.trafficgovernor.IRPCServiceTrafficService;
import org.apache.bifromq.dist.client.IDistClient;
import org.apache.bifromq.dist.server.IDistServer;
import org.apache.bifromq.dist.worker.IDistWorker;
@@ -66,6 +54,18 @@ import org.apache.bifromq.inbox.server.IInboxServer;
import org.apache.bifromq.inbox.store.IInboxStore;
import org.apache.bifromq.mqtt.IMQTTBroker;
import org.apache.bifromq.mqtt.inbox.IMqttBrokerClient;
+import org.apache.bifromq.plugin.authprovider.IAuthProvider;
+import org.apache.bifromq.plugin.clientbalancer.IClientBalancer;
+import org.apache.bifromq.plugin.eventcollector.Event;
+import org.apache.bifromq.plugin.eventcollector.IEventCollector;
+import org.apache.bifromq.plugin.resourcethrottler.IResourceThrottler;
+import org.apache.bifromq.plugin.settingprovider.ISettingProvider;
+import org.apache.bifromq.plugin.settingprovider.Setting;
+import org.apache.bifromq.plugin.subbroker.ISubBrokerManager;
+import org.apache.bifromq.plugin.subbroker.SubBrokerManager;
+import org.apache.bifromq.retain.client.IRetainClient;
+import org.apache.bifromq.retain.server.IRetainServer;
+import org.apache.bifromq.retain.store.IRetainStore;
import org.apache.bifromq.sessiondict.client.ISessionDictClient;
import org.apache.bifromq.sessiondict.rpc.proto.KillReply;
import org.apache.bifromq.sessiondict.rpc.proto.ServerRedirection;
@@ -247,6 +247,7 @@ public abstract class MQTTTest {
.setDataEngineConfigurator(new InMemKVEngineConfigurator())
.setWalEngineConfigurator(new InMemKVEngineConfigurator()))
.subBrokerManager(inboxBrokerMgr)
+ .settingProvider(settingProvider)
.build();
distServer = IDistServer.builder()
.rpcServerBuilder(rpcServerBuilder)
@@ -254,7 +255,6 @@ public abstract class MQTTTest {
.settingProvider(settingProvider)
.eventCollector(eventCollector)
.build();
-
mqttBroker = IMQTTBroker.builder()
.rpcServerBuilder(rpcServerBuilder)
.mqttBossELGThreads(1)
diff --git a/bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/EventType.java b/bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/EventType.java
index a8990d69..08eade1b 100644
--- a/bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/EventType.java
+++ b/bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/EventType.java
@@ -14,7 +14,7 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
package org.apache.bifromq.plugin.eventcollector;
@@ -98,16 +98,14 @@ public enum EventType {
DISTED,
DIST_ERROR,
DELIVER_ERROR,
- DELIVER_NO_INBOX, // deprecated
+ PERSISTENT_FANOUT_THROTTLED,
+ PERSISTENT_FANOUT_BYTES_THROTTLED,
+ GROUP_FANOUT_THROTTLED,
DELIVERED,
MATCHED,
MATCH_ERROR,
UNMATCHED,
UNMATCH_ERROR,
- SUBSCRIBED,
- SUBSCRIBE_ERROR,
- UNSUBSCRIBED,
- UNSUBSCRIBED_ERROR,
// inbox service
OVERFLOWED,
diff --git a/bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/DeliverNoInbox.java b/bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/DeliverNoInbox.java
deleted file mode 100644
index 2b32e1cd..00000000
--- a/bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/DeliverNoInbox.java
+++ /dev/null
@@ -1,55 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.bifromq.plugin.eventcollector.distservice;
-
-import org.apache.bifromq.plugin.eventcollector.Event;
-import org.apache.bifromq.plugin.eventcollector.EventType;
-import org.apache.bifromq.type.MatchInfo;
-import org.apache.bifromq.type.TopicMessagePack;
-import lombok.Getter;
-import lombok.Setter;
-import lombok.ToString;
-import lombok.experimental.Accessors;
-
-@Getter
-@Setter
-@Accessors(fluent = true, chain = true)
-@ToString(callSuper = true)
-@Deprecated(since = "4.0.0", forRemoval = true)
-public final class DeliverNoInbox extends Event<DeliverNoInbox> {
- private int brokerId;
- private String delivererKey;
- private MatchInfo subInfo;
- private TopicMessagePack messages;
-
- @Override
- public EventType type() {
- return EventType.DELIVER_NO_INBOX;
- }
-
- @Override
- public void clone(DeliverNoInbox orig) {
- super.clone(orig);
- this.brokerId = orig.brokerId;
- this.delivererKey = orig.delivererKey;
- this.subInfo = orig.subInfo;
- this.messages = orig.messages;
- }
-}
diff --git a/bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/Unsubscribed.java b/bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/GroupFanoutThrottled.java
similarity index 73%
rename from bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/Unsubscribed.java
rename to bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/GroupFanoutThrottled.java
index 86095007..f4b2b8e1 100644
--- a/bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/Unsubscribed.java
+++ b/bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/GroupFanoutThrottled.java
@@ -19,38 +19,32 @@
package org.apache.bifromq.plugin.eventcollector.distservice;
-import org.apache.bifromq.plugin.eventcollector.Event;
-import org.apache.bifromq.plugin.eventcollector.EventType;
import lombok.Getter;
import lombok.Setter;
import lombok.ToString;
import lombok.experimental.Accessors;
+import org.apache.bifromq.plugin.eventcollector.Event;
+import org.apache.bifromq.plugin.eventcollector.EventType;
@Getter
@Setter
@Accessors(fluent = true, chain = true)
@ToString(callSuper = true)
-public final class Unsubscribed extends Event<Unsubscribed> {
- private long reqId;
- private String topicFilter;
+public final class GroupFanoutThrottled extends Event<GroupFanoutThrottled> {
private String tenantId;
- private String inboxId;
- private int subBrokerId;
- private String delivererKey;
+ private String topic;
+ private int maxCount;
@Override
public EventType type() {
- return EventType.UNSUBSCRIBED;
+ return EventType.GROUP_FANOUT_THROTTLED;
}
@Override
- public void clone(Unsubscribed orig) {
+ public void clone(GroupFanoutThrottled orig) {
super.clone(orig);
- this.reqId = orig.reqId;
- this.topicFilter = orig.topicFilter;
this.tenantId = orig.tenantId;
- this.inboxId = orig.inboxId;
- this.subBrokerId = orig.subBrokerId;
- this.delivererKey = orig.delivererKey;
+ this.topic = orig.topic;
+ this.maxCount = orig.maxCount;
}
}
diff --git a/bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/UnsubscribeError.java b/bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/PersistentFanoutBytesThrottled.java
similarity index 69%
rename from bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/UnsubscribeError.java
rename to bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/PersistentFanoutBytesThrottled.java
index fc4eea45..7c686ae2 100644
--- a/bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/UnsubscribeError.java
+++ b/bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/PersistentFanoutBytesThrottled.java
@@ -14,44 +14,37 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
package org.apache.bifromq.plugin.eventcollector.distservice;
-import org.apache.bifromq.plugin.eventcollector.Event;
-import org.apache.bifromq.plugin.eventcollector.EventType;
import lombok.Getter;
import lombok.Setter;
import lombok.ToString;
import lombok.experimental.Accessors;
+import org.apache.bifromq.plugin.eventcollector.Event;
+import org.apache.bifromq.plugin.eventcollector.EventType;
@Getter
@Setter
@Accessors(fluent = true, chain = true)
@ToString(callSuper = true)
-@Deprecated(since = "3.1.1", forRemoval = true)
-public final class UnsubscribeError extends Event<UnsubscribeError> {
- private long reqId;
- private String topicFilter;
+public final class PersistentFanoutBytesThrottled extends Event<PersistentFanoutBytesThrottled> {
private String tenantId;
- private String inboxId;
- private int subBrokerId;
- private String delivererKey;
+ private String topic;
+ private long maxBytes;
@Override
public EventType type() {
- return EventType.UNSUBSCRIBED_ERROR;
+ return EventType.PERSISTENT_FANOUT_BYTES_THROTTLED;
}
@Override
- public void clone(UnsubscribeError orig) {
+ public void clone(PersistentFanoutBytesThrottled orig) {
super.clone(orig);
- this.reqId = orig.reqId;
- this.topicFilter = orig.topicFilter;
this.tenantId = orig.tenantId;
- this.inboxId = orig.inboxId;
- this.subBrokerId = orig.subBrokerId;
- this.delivererKey = orig.delivererKey;
+ this.topic = orig.topic;
+ this.maxBytes = orig.maxBytes;
}
}
diff --git a/bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/Subscribed.java b/bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/PersistentFanoutThrottled.java
similarity index 68%
rename from bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/Subscribed.java
rename to bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/PersistentFanoutThrottled.java
index a2c97323..bf5b95ce 100644
--- a/bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/Subscribed.java
+++ b/bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/PersistentFanoutThrottled.java
@@ -14,46 +14,37 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
package org.apache.bifromq.plugin.eventcollector.distservice;
-import org.apache.bifromq.plugin.eventcollector.Event;
-import org.apache.bifromq.plugin.eventcollector.EventType;
-import org.apache.bifromq.type.QoS;
import lombok.Getter;
import lombok.Setter;
import lombok.ToString;
import lombok.experimental.Accessors;
+import org.apache.bifromq.plugin.eventcollector.Event;
+import org.apache.bifromq.plugin.eventcollector.EventType;
@Getter
@Setter
@Accessors(fluent = true, chain = true)
@ToString(callSuper = true)
-public final class Subscribed extends Event<Subscribed> {
- private long reqId;
- private String topicFilter;
- private QoS qos;
+public final class PersistentFanoutThrottled extends Event<PersistentFanoutThrottled> {
private String tenantId;
- private String inboxId;
- private int subBrokerId;
- private String delivererKey;
+ private String topic;
+ private int maxCount;
@Override
public EventType type() {
- return EventType.SUBSCRIBED;
+ return EventType.PERSISTENT_FANOUT_THROTTLED;
}
@Override
- public void clone(Subscribed orig) {
+ public void clone(PersistentFanoutThrottled orig) {
super.clone(orig);
- this.reqId = orig.reqId;
- this.topicFilter = orig.topicFilter;
- this.qos = orig.qos;
this.tenantId = orig.tenantId;
- this.inboxId = orig.inboxId;
- this.subBrokerId = orig.subBrokerId;
- this.delivererKey = orig.delivererKey;
+ this.topic = orig.topic;
+ this.maxCount = orig.maxCount;
}
}
diff --git a/bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/SubscribeError.java b/bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/SubscribeError.java
deleted file mode 100644
index f4af7614..00000000
--- a/bifromq-plugin/bifromq-plugin-event-collector/src/main/java/org/apache/bifromq/plugin/eventcollector/distservice/SubscribeError.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.bifromq.plugin.eventcollector.distservice;
-
-import org.apache.bifromq.plugin.eventcollector.Event;
-import org.apache.bifromq.plugin.eventcollector.EventType;
-import org.apache.bifromq.type.QoS;
-import lombok.Getter;
-import lombok.Setter;
-import lombok.ToString;
-import lombok.experimental.Accessors;
-
-@Getter
-@Setter
-@Accessors(fluent = true, chain = true)
-@ToString(callSuper = true)
-@Deprecated(since = "3.1.1", forRemoval = true)
-public final class SubscribeError extends Event<SubscribeError> {
- private long reqId;
- private String topicFilter;
- private QoS qos;
- private String tenantId;
- private String inboxId;
- private int subBrokerId;
- private String delivererKey;
-
- @Override
- public EventType type() {
- return EventType.SUBSCRIBE_ERROR;
- }
-
- @Override
- public void clone(SubscribeError orig) {
- super.clone(orig);
- this.reqId = orig.reqId;
- this.topicFilter = orig.topicFilter;
- this.qos = orig.qos;
- this.tenantId = orig.tenantId;
- this.inboxId = orig.inboxId;
- this.subBrokerId = orig.subBrokerId;
- this.delivererKey = orig.delivererKey;
- }
-}
diff --git a/bifromq-plugin/bifromq-plugin-setting-provider/src/main/java/org/apache/bifromq/plugin/settingprovider/Setting.java b/bifromq-plugin/bifromq-plugin-setting-provider/src/main/java/org/apache/bifromq/plugin/settingprovider/Setting.java
index 3125e318..a5961918 100644
--- a/bifromq-plugin/bifromq-plugin-setting-provider/src/main/java/org/apache/bifromq/plugin/settingprovider/Setting.java
+++ b/bifromq-plugin/bifromq-plugin-setting-provider/src/main/java/org/apache/bifromq/plugin/settingprovider/Setting.java
@@ -14,7 +14,7 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
package org.apache.bifromq.plugin.settingprovider;
@@ -51,6 +51,9 @@ public enum Setting {
MaxResendTimes(Integer.class, val -> (int) val >= 0, 3),
ResendTimeoutSeconds(Integer.class, val -> (int) val > 0, 10),
MaxTopicFiltersPerSub(Integer.class, val -> (int) val > 0 && (int) val <= 100, 10),
+ MaxGroupFanout(Integer.class, val -> (int) val > 0, 100),
+ MaxPersistentFanout(Integer.class, val -> (int) val > 0, Integer.MAX_VALUE),
+ MaxPersistentFanoutBytes(Long.class, val -> (long) val > 0, Long.MAX_VALUE),
MaxSessionExpirySeconds(Integer.class, val -> (int) val > 0 && Integer.compareUnsigned((int) val, 0xFFFFFFFF) <= 0,
24 * 60 * 60),
MinSessionExpirySeconds(Integer.class, (val, tenantId) -> {
diff --git a/bifromq-retain/bifromq-retain-server/src/main/java/org/apache/bifromq/retain/server/scheduler/BatchMatchCall.java b/bifromq-retain/bifromq-retain-server/src/main/java/org/apache/bifromq/retain/server/scheduler/BatchMatchCall.java
index 140aaa7d..22ba7079 100644
--- a/bifromq-retain/bifromq-retain-server/src/main/java/org/apache/bifromq/retain/server/scheduler/BatchMatchCall.java
+++ b/bifromq-retain/bifromq-retain-server/src/main/java/org/apache/bifromq/retain/server/scheduler/BatchMatchCall.java
@@ -26,6 +26,16 @@ import static org.apache.bifromq.retain.server.scheduler.BatchMatchCallHelper.se
import static org.apache.bifromq.retain.server.scheduler.MatchCallRangeRouter.rangeLookup;
import static org.apache.bifromq.util.TopicUtil.isWildcardTopicFilter;
+import java.util.ArrayDeque;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import lombok.extern.slf4j.Slf4j;
import org.apache.bifromq.basehlc.HLC;
import org.apache.bifromq.basekv.client.IBaseKVStoreClient;
import org.apache.bifromq.basekv.client.KVRangeSetting;
@@ -47,26 +57,16 @@ import org.apache.bifromq.retain.rpc.proto.MatchParam;
import org.apache.bifromq.retain.rpc.proto.MatchReply;
import org.apache.bifromq.retain.rpc.proto.MatchResult;
import org.apache.bifromq.retain.rpc.proto.RetainServiceROCoProcInput;
-import java.util.ArrayDeque;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.NavigableMap;
-import java.util.Queue;
-import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-import lombok.extern.slf4j.Slf4j;
@Slf4j
class BatchMatchCall implements IBatchCall<MatchRetainedRequest, MatchRetainedResult, MatchCallBatcherKey> {
private final MatchCallBatcherKey batcherKey;
private final IBaseKVStoreClient retainStoreClient;
private final ISettingProvider settingProvider;
- private final Queue<ICallTask<MatchRetainedRequest, MatchRetainedResult, MatchCallBatcherKey>> tasks = new ArrayDeque<>(
- 128);
- private Set<String> nonWildcardTopicFilters = new HashSet<>(128);
- private Set<String> wildcardTopicFilters = new HashSet<>(128);
+ private final Queue<ICallTask<MatchRetainedRequest, MatchRetainedResult, MatchCallBatcherKey>> tasks =
+ new ArrayDeque<>(128);
+ private final Set<String> nonWildcardTopicFilters = new HashSet<>(128);
+ private final Set<String> wildcardTopicFilters = new HashSet<>(128);
BatchMatchCall(MatchCallBatcherKey batcherKey, IBaseKVStoreClient retainStoreClient,
@@ -88,8 +88,8 @@ class BatchMatchCall implements IBatchCall<MatchRetainedRequest, MatchRetainedRe
@Override
public void reset() {
- nonWildcardTopicFilters = new HashSet<>(128);
- wildcardTopicFilters = new HashSet<>(128);
+ nonWildcardTopicFilters.clear();
+ wildcardTopicFilters.clear();
}
@Override
diff --git a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/ITenantRouteMatcher.java b/bifromq-sysprops/src/main/java/org/apache/bifromq/sysprops/props/DistCachedRoutesFanoutCheckIntervalSeconds.java
similarity index 54%
copy from bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/ITenantRouteMatcher.java
copy to bifromq-sysprops/src/main/java/org/apache/bifromq/sysprops/props/DistCachedRoutesFanoutCheckIntervalSeconds.java
index 191155d9..a3e49f17 100644
--- a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/cache/ITenantRouteMatcher.java
+++ b/bifromq-sysprops/src/main/java/org/apache/bifromq/sysprops/props/DistCachedRoutesFanoutCheckIntervalSeconds.java
@@ -14,24 +14,21 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
-package org.apache.bifromq.dist.worker.cache;
+package org.apache.bifromq.sysprops.props;
-import org.apache.bifromq.dist.worker.schema.Matching;
-import java.util.Map;
-import java.util.Set;
+import org.apache.bifromq.sysprops.BifroMQSysProp;
+import org.apache.bifromq.sysprops.parser.IntegerParser;
/**
- * The interface for tenant route matcher.
+ * The system property for the interval seconds of checking cached routes against fanout related settings.
*/
-public interface ITenantRouteMatcher {
- /**
- * Match the topic within the given boundary.
- *
- * @param topics the topics to match
- * @return the result
- */
- Map<String, Set<Matching>> matchAll(Set<String> topics);
+public final class DistCachedRoutesFanoutCheckIntervalSeconds extends BifroMQSysProp<Integer, IntegerParser> {
+ public static final DistCachedRoutesFanoutCheckIntervalSeconds INSTANCE = new DistCachedRoutesFanoutCheckIntervalSeconds();
+
+ private DistCachedRoutesFanoutCheckIntervalSeconds() {
+ super("dist_worker_cache_fanout_check_seconds", 5, IntegerParser.POSITIVE);
+ }
}
diff --git a/build/build-bifromq-starter/src/main/java/org/apache/bifromq/starter/module/DistServiceModule.java b/build/build-bifromq-starter/src/main/java/org/apache/bifromq/starter/module/DistServiceModule.java
index 5a82f765..675b9bfd 100644
--- a/build/build-bifromq-starter/src/main/java/org/apache/bifromq/starter/module/DistServiceModule.java
+++ b/build/build-bifromq-starter/src/main/java/org/apache/bifromq/starter/module/DistServiceModule.java
@@ -14,7 +14,7 @@
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
- * under the License.
+ * under the License.
*/
package org.apache.bifromq.starter.module;
@@ -133,6 +133,7 @@ public class DistServiceModule extends AbstractModule {
.balancerRetryDelay(Duration.ofMillis(workerConfig.getBalanceConfig().getRetryDelayInMS()))
.balancerFactoryConfig(workerConfig.getBalanceConfig().getBalancers())
.subBrokerManager(injector.getInstance(ISubBrokerManager.class))
+ .settingProvider(injector.getInstance(SettingProviderManager.class))
.loadEstimateWindow(Duration.ofSeconds(DistWorkerLoadEstimationWindowSeconds.INSTANCE.get()))
.attributes(workerConfig.getAttributes())
.build());
diff --git a/pom.xml b/pom.xml
index 7c550cf9..7bf0e517 100644
--- a/pom.xml
+++ b/pom.xml
@@ -108,7 +108,7 @@
<apache.commons.cli.version>1.5.0</apache.commons.cli.version>
<apache.commons.collections.version>4.4</apache.commons.collections.version>
<bouncycastle.version>1.78.1</bouncycastle.version>
- <caffeine.version>3.1.8</caffeine.version>
+ <caffeine.version>3.2.2</caffeine.version>
<guice.version>7.0.0</guice.version>
<grpc.version>1.72.0</grpc.version>
<guava.version>33.2.1-jre</guava.version>