[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

Author: popduke

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,34 +26,35 @@
 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> {
     private final IDistClient distClient;
     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,15 +77,20 @@ public void add(ICallTask<DeliveryCall, DeliveryCallResult, DelivererKey> callTa
     @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);
                 matchInfos.forEach(packBuilder::addMatchInfo);
                 packageBuilder.addPack(packBuilder.build());
             });
             requestBuilder.putPackage(tenantId, packageBuilder.build());
-        });
+            itr.remove();
+        }
         DeliveryRequest request = requestBuilder.build();
         return execute(request);
     }

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.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 @@ public void add(ICallTask<PubRequest, PubResult, PubCallBatcherKey> callTask) {
     @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);
     }

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 @@ public void add(ICallTask<TenantPubRequest, DistServerCallResult, DistServerCall
 
     @Override
     public void reset() {
-        batch = new HashMap<>(128);
+        batch.clear();
     }
 
     @Override

bifromq-dist/bifromq-dist-server/src/test/java/org/apache/bifromq/dist/server/DistServiceTest.java

@@ -24,6 +24,10 @@
 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.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 void setup() {
             .bgTaskExecutor(bgTaskExecutor)
             .storeOptions(kvRangeStoreOptions)
             .subBrokerManager(subBrokerMgr)
+            .settingProvider(settingProvider)
             .build();
         distServer = IDistServer.builder()
             .rpcServerBuilder(rpcServerBuilder)

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;

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.

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.

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>