Outbound rate limiting for persistent APPLICATION clients

application/src/main/java/org/thingsboard/mqtt/broker/config/ApplicationPersistedMsgsRateLimitsConfiguration.java

@@ -0,0 +1,30 @@
+/**
+ * Copyright © 2016-2026 The Thingsboard Authors
+ *
+ * Licensed 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.thingsboard.mqtt.broker.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+@ConfigurationProperties(prefix = "mqtt.rate-limits.application-persisted-messages")
+@Data
+public class ApplicationPersistedMsgsRateLimitsConfiguration {
+
+    private boolean enabled;
+    private String clientConfig;
+
+}

application/src/main/java/org/thingsboard/mqtt/broker/service/limits/RateLimitService.java

@@ -46,6 +46,13 @@ public interface RateLimitService {
 
     boolean isDevicePersistedMsgsLimitEnabled();
 
+    /**
+     * Non-blocking token consume for outbound persisted messages to an APPLICATION client (throttling, not dropping)
+     */
+    boolean tryConsumeApplicationPersistedMsgs(String clientId);
+
+    boolean isApplicationPersistedMsgsRateLimitEnabled();
+
     long tryConsumeTotalMsgs(long limit);
 
     boolean isTotalMsgsLimitEnabled();

application/src/main/java/org/thingsboard/mqtt/broker/service/limits/RateLimitServiceImpl.java

@@ -24,6 +24,7 @@
 import org.thingsboard.mqtt.broker.common.data.ClientSessionInfo;
 import org.thingsboard.mqtt.broker.common.data.SessionInfo;
 import org.thingsboard.mqtt.broker.common.util.TbRateLimits;
+import org.thingsboard.mqtt.broker.config.ApplicationPersistedMsgsRateLimitsConfiguration;
 import org.thingsboard.mqtt.broker.config.ClientsLimitProperties;
 import org.thingsboard.mqtt.broker.config.DevicePersistedMsgsRateLimitsConfiguration;
 import org.thingsboard.mqtt.broker.config.IncomingRateLimitsConfiguration;
@@ -44,13 +45,16 @@ public class RateLimitServiceImpl implements RateLimitService {
     private final OutgoingRateLimitsConfiguration outgoingRateLimitsConfiguration;
     private final TotalMsgsRateLimitsConfiguration totalMsgsRateLimitsConfiguration;
     private final DevicePersistedMsgsRateLimitsConfiguration devicePersistedMsgsRateLimitsConfiguration;
+    private final ApplicationPersistedMsgsRateLimitsConfiguration applicationPersistedMsgsRateLimitsConfiguration;
     private final RateLimitCacheService rateLimitCacheService;
     private final ClientsLimitProperties clientsLimitProperties;
 
     @Getter
     private ConcurrentMap<String, TbRateLimits> incomingPublishClientLimits;
     @Getter
     private ConcurrentMap<String, TbRateLimits> outgoingPublishClientLimits;
+    @Getter
+    private ConcurrentMap<String, TbRateLimits> applicationPersistedMsgClientLimits;
 
     @PostConstruct
     public void init() {
@@ -60,6 +64,9 @@ public void init() {
         if (outgoingRateLimitsConfiguration.isEnabled()) {
             outgoingPublishClientLimits = new ConcurrentHashMap<>();
         }
+        if (applicationPersistedMsgsRateLimitsConfiguration.isEnabled()) {
+            applicationPersistedMsgClientLimits = new ConcurrentHashMap<>();
+        }
     }
 
     @Override
@@ -100,6 +107,9 @@ public void remove(String clientId) {
             if (outgoingPublishClientLimits != null) {
                 outgoingPublishClientLimits.remove(clientId);
             }
+            if (applicationPersistedMsgClientLimits != null) {
+                applicationPersistedMsgClientLimits.remove(clientId);
+            }
         }
     }
 
@@ -175,6 +185,21 @@ public boolean isDevicePersistedMsgsLimitEnabled() {
         return devicePersistedMsgsRateLimitsConfiguration.isEnabled();
     }
 
+    @Override
+    public boolean tryConsumeApplicationPersistedMsgs(String clientId) {
+        if (!applicationPersistedMsgsRateLimitsConfiguration.isEnabled()) {
+            return true;
+        }
+        TbRateLimits rateLimits = applicationPersistedMsgClientLimits.computeIfAbsent(
+                clientId, id -> new TbRateLimits(applicationPersistedMsgsRateLimitsConfiguration.getClientConfig()));
+        return rateLimits.tryConsume();
+    }
+
+    @Override
+    public boolean isApplicationPersistedMsgsRateLimitEnabled() {
+        return applicationPersistedMsgsRateLimitsConfiguration.isEnabled();
+    }
+
     @Override
     public long tryConsumeTotalMsgs(long limit) {
         return rateLimitCacheService.tryConsumeTotalMsgs(limit);

application/src/main/java/org/thingsboard/mqtt/broker/service/mqtt/persistence/application/ApplicationPersistenceProcessorImpl.java

@@ -27,6 +27,7 @@
 import org.thingsboard.mqtt.broker.actors.client.messages.mqtt.MqttDisconnectMsg;
 import org.thingsboard.mqtt.broker.actors.client.state.ClientActorStateInfo;
 import org.thingsboard.mqtt.broker.adaptor.ProtoConverter;
+import org.thingsboard.mqtt.broker.common.data.PersistedPacketType;
 import org.thingsboard.mqtt.broker.common.data.mqtt.MsgExpiryResult;
 import org.thingsboard.mqtt.broker.common.util.ThingsBoardExecutors;
 import org.thingsboard.mqtt.broker.gen.queue.PublishMsgProto;
@@ -38,6 +39,7 @@
 import org.thingsboard.mqtt.broker.queue.provider.ApplicationPersistenceMsgQueueFactory;
 import org.thingsboard.mqtt.broker.service.analysis.ClientLogger;
 import org.thingsboard.mqtt.broker.service.historical.stats.TbMessageStatsReportClient;
+import org.thingsboard.mqtt.broker.service.limits.RateLimitService;
 import org.thingsboard.mqtt.broker.service.mqtt.MqttMsgDeliveryService;
 import org.thingsboard.mqtt.broker.service.mqtt.PublishMsg;
 import org.thingsboard.mqtt.broker.service.mqtt.persistence.application.data.ApplicationMainProcessingState;
@@ -81,6 +83,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
 import java.util.stream.Collectors;
 
 @Service
@@ -112,6 +115,7 @@ public class ApplicationPersistenceProcessorImpl implements ApplicationPersisten
     private final ApplicationClientHelperService appClientHelperService;
     private final AppMsgDeliveryStrategy appMsgDeliveryStrategy;
     private final TbMessageStatsReportClient tbMessageStatsReportClient;
+    private final RateLimitService rateLimitService;
     private final boolean isDebugEnabled = log.isDebugEnabled();
 
     @Value("${queue.application-persisted-msg.poll-interval}")
@@ -453,6 +457,7 @@ private ApplicationPubRelMsgCtx processMainPack(ApplicationPubRelMsgCtx pubRelMs
 
         List<PersistedMsg> messagesToDeliver = buildMessagesToDeliver(pubRelMsgCtx, clientSessionCtx, persistedMsgCtx, messages, null);
         submitStrategy.init(messagesToDeliver);
+        throttleDelivery(clientId, messagesToDeliver, () -> isClientSessionActive(sessionId, clientState));
 
         if (isDebugEnabled) {
             log.debug("[{}] Starting main pack, {} messages to deliver", clientId, messagesToDeliver.size());
@@ -558,6 +563,7 @@ private ApplicationPubRelMsgCtx processSharedPack(ApplicationPubRelMsgCtx pubRel
 
         List<PersistedMsg> messagesToDeliver = buildMessagesToDeliver(pubRelMsgCtx, clientSessionCtx, persistedMsgCtx, messages, subscription);
         submitStrategy.init(messagesToDeliver);
+        throttleDelivery(clientId, messagesToDeliver, () -> isJobActive(job));
 
         if (log.isTraceEnabled()) {
             log.trace("[{}] Starting shared subscription pack, {} messages to deliver", clientId, messagesToDeliver.size());
@@ -731,6 +737,46 @@ private void deliverMessages(ApplicationSubmitStrategy submitStrategy, ClientSes
         appMsgDeliveryStrategy.process(submitStrategy, clientSessionCtx);
     }
 
+    void throttleDelivery(String clientId, List<PersistedMsg> messagesToDeliver, BooleanSupplier isActive) {
+        if (!rateLimitService.isApplicationPersistedMsgsRateLimitEnabled()) {
+            return;
+        }
+        int remaining = countPublishMsgs(messagesToDeliver);
+        boolean throttled = false;
+        while (remaining > 0) {
+            if (!isActive.getAsBoolean()) {
+                return;
+            }
+            if (rateLimitService.tryConsumeApplicationPersistedMsgs(clientId)) {
+                remaining--;
+            } else {
+                if (!throttled) {
+                    throttled = true;
+                    if (isDebugEnabled) {
+                        log.debug("[{}] Outbound rate limit reached; pacing delivery of {} remaining message(s) in this pack", clientId, remaining);
+                    }
+                }
+                // Reuse the Kafka poll interval as the throttle back-off granularity (intentionally no separate tunable).
+                try {
+                    Thread.sleep(pollDuration);
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    return;
+                }
+            }
+        }
+    }
+
+    private static int countPublishMsgs(List<PersistedMsg> messagesToDeliver) {
+        int count = 0;
+        for (PersistedMsg msg : messagesToDeliver) {
+            if (PersistedPacketType.PUBLISH == msg.getPacketType()) {
+                count++;
+            }
+        }
+        return count;
+    }
+
     private void processPubAckInSharedCtx(String clientId, int packetId, String format) {
         Set<ApplicationSharedSubscriptionCtx> contexts = sharedPackProcessingCtxMap.get(clientId);
         if (CollectionUtils.isEmpty(contexts)) {

application/src/main/resources/thingsboard-mqtt-broker.yml

@@ -1028,6 +1028,19 @@ mqtt:
       # Limits the count of Device clients persisted messages per time interval (in s). Comma-separated list of limit:seconds pairs.
       # Example: 100 messages per second or 1000 messages per minute.
       config: "${MQTT_DEVICE_PERSISTED_MSGS_RATE_LIMITS_CONFIG:100:1,1000:60}"
+    application-persisted-messages:
+      # Enables or disables per-client outbound rate limits for persistent APPLICATION subscribers.
+      # Throttles (paces) delivery without dropping messages: the Kafka backlog is retained and
+      # replayed at the configured rate. Applies to the main and shared-subscription consumers.
+      enabled: "${MQTT_APPLICATION_PERSISTED_MSGS_RATE_LIMITS_ENABLED:false}"
+      # Limits the count of outgoing persisted messages per APPLICATION client per time interval (in s).
+      # Comma-separated list of limit:seconds pairs.
+      # Example: 100 messages per second or 5000 messages per minute.
+      # Note: for shared subscriptions this paces a shared Kafka consumer group. Avoid a rate so low that
+      # draining one polled pack (up to max.poll.records) takes longer than max.poll.interval.ms (default 300s),
+      # otherwise the shared-subscription consumer may be evicted and its group rebalanced. The per-client
+      # (non-shared) APPLICATION consumer uses manual partition assignment and is not affected.
+      client-config: "${MQTT_APPLICATION_PERSISTED_MSGS_RATE_LIMITS_CLIENT_CONFIG:100:1,5000:60}"
   # Total limit of sessions (connected + disconnected) stored on the broker, applied collectively across the cluster, not per node.
   # For example, when set to 1000, the entire cluster can store 1000 sessions in total. This is a soft limit, meaning slightly more sessions may be stored.
   # Defaults to 0, which disables the limit.

application/src/test/java/org/thingsboard/mqtt/broker/service/limits/RateLimitServiceImplTest.java

@@ -29,6 +29,7 @@
 import org.thingsboard.mqtt.broker.common.data.ClientType;
 import org.thingsboard.mqtt.broker.common.data.SessionInfo;
 import org.thingsboard.mqtt.broker.common.util.TbRateLimits;
+import org.thingsboard.mqtt.broker.config.ApplicationPersistedMsgsRateLimitsConfiguration;
 import org.thingsboard.mqtt.broker.config.ClientsLimitProperties;
 import org.thingsboard.mqtt.broker.config.DevicePersistedMsgsRateLimitsConfiguration;
 import org.thingsboard.mqtt.broker.config.IncomingRateLimitsConfiguration;
@@ -55,6 +56,8 @@ public class RateLimitServiceImplTest {
     @MockitoBean
     DevicePersistedMsgsRateLimitsConfiguration devicePersistedMsgsRateLimitsConfiguration;
     @MockitoBean
+    ApplicationPersistedMsgsRateLimitsConfiguration applicationPersistedMsgsRateLimitsConfiguration;
+    @MockitoBean
     TotalMsgsRateLimitsConfiguration totalMsgsRateLimitsConfiguration;
     @MockitoBean
     RateLimitCacheService rateLimitCacheService;
@@ -69,12 +72,14 @@ public void setUp() throws Exception {
         when(incomingRateLimitsConfiguration.isEnabled()).thenReturn(true);
         when(outgoingRateLimitsConfiguration.isEnabled()).thenReturn(true);
         when(devicePersistedMsgsRateLimitsConfiguration.isEnabled()).thenReturn(true);
+        when(applicationPersistedMsgsRateLimitsConfiguration.isEnabled()).thenReturn(true);
         when(totalMsgsRateLimitsConfiguration.isEnabled()).thenReturn(true);
 
         rateLimitService.init();
 
         rateLimitService.getIncomingPublishClientLimits().put(CLIENT_ID, new TbRateLimits("1:1")); // limit 1 per 1 second
         rateLimitService.getOutgoingPublishClientLimits().put(CLIENT_ID, new TbRateLimits("1:1")); // limit 1 per 1 second
+        rateLimitService.getApplicationPersistedMsgClientLimits().put(CLIENT_ID, new TbRateLimits("1:1")); // limit 1 per 1 second
     }
 
     @After
@@ -157,13 +162,15 @@ public void givenOneClient_whenRemoveIt_thenSuccess() {
         rateLimitService.remove(CLIENT_ID);
         assertEquals(0, rateLimitService.getIncomingPublishClientLimits().size());
         assertEquals(0, rateLimitService.getOutgoingPublishClientLimits().size());
+        assertEquals(0, rateLimitService.getApplicationPersistedMsgClientLimits().size());
     }
 
     @Test
     public void givenOneClient_whenRemoveNull_thenSuccess() {
         rateLimitService.remove(null);
         assertEquals(1, rateLimitService.getIncomingPublishClientLimits().size());
         assertEquals(1, rateLimitService.getOutgoingPublishClientLimits().size());
+        assertEquals(1, rateLimitService.getApplicationPersistedMsgClientLimits().size());
     }
 
     @Test
@@ -303,4 +310,38 @@ public void givenTokensAvailable_whenTryConsumeTotalMsgs_thenSuccess() {
         long tokens = rateLimitService.tryConsumeTotalMsgs(10L);
         assertEquals(10L, tokens);
     }
+
+    @Test
+    public void givenAppPersistedMsgsRateLimitsDisabled_whenTryConsume_thenAlwaysTrue() {
+        when(applicationPersistedMsgsRateLimitsConfiguration.isEnabled()).thenReturn(false);
+
+        Assert.assertTrue(rateLimitService.tryConsumeApplicationPersistedMsgs(CLIENT_ID));
+        Assert.assertTrue(rateLimitService.tryConsumeApplicationPersistedMsgs(CLIENT_ID));
+    }
+
+    @Test
+    public void givenAppPersistedMsgsRateLimitsEnabled_whenTryConsume_thenGetExpectedResult() {
+        when(applicationPersistedMsgsRateLimitsConfiguration.isEnabled()).thenReturn(true);
+
+        Assert.assertTrue(rateLimitService.tryConsumeApplicationPersistedMsgs(CLIENT_ID));
+        Assert.assertFalse(rateLimitService.tryConsumeApplicationPersistedMsgs(CLIENT_ID));
+        Assert.assertFalse(rateLimitService.tryConsumeApplicationPersistedMsgs(CLIENT_ID));
+    }
+
+    @Test
+    public void givenTwoClients_whenTryConsume_thenBucketsAreIndependent() {
+        when(applicationPersistedMsgsRateLimitsConfiguration.isEnabled()).thenReturn(true);
+        rateLimitService.getApplicationPersistedMsgClientLimits().put("other", new TbRateLimits("1:1"));
+
+        Assert.assertTrue(rateLimitService.tryConsumeApplicationPersistedMsgs(CLIENT_ID));
+        Assert.assertTrue(rateLimitService.tryConsumeApplicationPersistedMsgs("other"));
+        Assert.assertFalse(rateLimitService.tryConsumeApplicationPersistedMsgs(CLIENT_ID));
+        Assert.assertFalse(rateLimitService.tryConsumeApplicationPersistedMsgs("other"));
+    }
+
+    @Test
+    public void givenEnabledConfig_whenIsApplicationPersistedMsgsRateLimitEnabled_thenTrue() {
+        when(applicationPersistedMsgsRateLimitsConfiguration.isEnabled()).thenReturn(true);
+        Assert.assertTrue(rateLimitService.isApplicationPersistedMsgsRateLimitEnabled());
+    }
 }