Outbound rate limiting for persistent APPLICATION clients

[dmytro-landiak]

Pull Request description

Closes #305.

Adds an opt-in, per-APPLICATION-client outbound rate limit that paces (throttles) delivery of persisted messages instead of dropping them. When a persistent APPLICATION subscriber reconnects after being offline, its Kafka backlog is currently replayed at full speed, which can overwhelm subscribers that auto-ACK and buffer internally. This throttles delivery to a configured rate, smoothing the replay into steady chunks. No data loss: offsets are committed only after acks and the backlog stays in Kafka — the per-client consumer thread is simply paced.

Scope of changes:

• New opt-in config mqtt.rate-limits.application-persisted-messages (enabled + client-config, same limit:seconds,... format as the other limiters; default off). Env vars MQTT_APPLICATION_PERSISTED_MSGS_RATE_LIMITS_ENABLED / ..._CLIENT_CONFIG.
• RateLimitService: per-client TbRateLimits bucket (reusing the existing bucket4j framework) with tryConsumeApplicationPersistedMsgs / isApplicationPersistedMsgsRateLimitEnabled; cleanup rides the existing remove(clientId) on disconnect.
• ApplicationPersistenceProcessorImpl: a session-aware throttle gate applied once per pack before delivery (in both the main and shared-subscription loops, before the retry loop so QoS retransmissions don't double-count; counts PUBLISH only).

Documentation notes: document the new setting/env vars and the shared-subscription caveat (avoid a rate so low that draining one polled pack takes longer than max.poll.interval.ms, or the shared consumer group may rebalance — the per-client main consumer uses manual partition assignment and is unaffected). A follow-up ticket will add a safeguard for that shared-subscription edge case.

General checklist

• You have reviewed the guidelines document.
• Labels that classify your pull request have been added.
• The milestone is specified and corresponds to fix version.
• Description references specific issue. ([Feature Request] Outbound rate limiting / throttling for MQTT subscribers (APPLICATION clients) #305)
• Description contains human-readable scope of changes.
• Description contains brief notes about what needs to be added to the documentation.
• No merge conflicts, commented blocks of code, code formatting issues.
• Changes are backward compatible or upgrade script is provided. (opt-in, default off)

Front-End feature checklist

• N/A — no front-end changes.

Back-End feature checklist

• Added corresponding unit and/or integration test(s). (RateLimitServiceImplTest, ApplicationPersistenceProcessorImplTest)
• If new dependency was added: the dependency tree is checked for conflicts. (no new dependency)

[dmytro-landiak]

Review summary

Reviewed 7 changed files in Outbound rate limiting for persistent APPLICATION clients. Left 7 comment(s) inline.

The feature is cleanly scoped and correctly opt-in (default off), reuses the existing TbRateLimits/bucket4j framework, and the no-data-loss reasoning holds up: tokens are consumed before delivery, offsets commit only after acks, and both the inactive-session and interrupt paths return without committing, so an un-throttled pack is simply re-polled. The shared-subscription rebalance caveat is already documented with a follow-up. Findings are mostly quality/consistency; the one correctness question is about message-expiry being evaluated before the throttle wait.

---

This review was auto-generated. Findings may contain errors — please verify before applying changes.

[dmytro-landiak]

Message expiry is evaluated once in buildPublishMessagesToDeliver (against System.currentTimeMillis() at build time), but throttleDelivery can then hold the pack here for seconds — or much longer at low limits — before deliverMessages runs. A message that was still within its expiry window when the pack was built could cross the boundary during the throttle wait and then be delivered anyway. Is that acceptable for MQTT message-expiry semantics, or should expiry be re-checked after throttling / just before the send?

[dmytro-landiak]

throttleDelivery acquires tokens for the whole pack before deliverMessages runs, so the pack is still handed off as a single burst once enough tokens accrue — the loop paces token acquisition, not the actual sends. With e.g. 100:1,5000:60 and a large polled pack (main up to 200, shared up to 500), the client sees nothing for a few seconds and then receives the whole pack at once rather than a steady drip. The bucket does bound the long-run average, so this is a reasonable simplification — just want to confirm it's intentional given the PR describes the goal as replaying at a steady rate. Was interleaving the gate with per-message delivery considered?

[dmytro-landiak]

Reusing queue.application-persisted-msg.poll-interval as the back-off granularity couples the throttle to a value that's really tuned for Kafka polling. If someone later changes the poll interval for Kafka reasons, the throttle back-off silently changes with it (and vice versa). The comment shows it's deliberate, which helps — just flagging that the coupling isn't visible from the call sites, and a small dedicated default might age better.

[dmytro-landiak]

The four new unit tests exercise throttleDelivery directly, but nothing verifies the wiring — that processMainPack/processSharedPack actually invoke it before the retry loop with the right isActive supplier. The existing processMainPack tests run with the limit mocked disabled (so the gate is a no-op), and the shared path has no throttling test at all. A test that drives these methods with the limit enabled and asserts tryConsumeApplicationPersistedMsgs is called would lock in the placement — it's easy to move this call to the wrong spot and not notice.

[dmytro-landiak]

This is now the third copy of the per-client bucket idiom — checkIncomingLimits (line 77) and checkOutgoingLimits (line 93) both do the same map.computeIfAbsent(clientId, id -> new TbRateLimits(config.getClientConfig())).tryConsume(). Would it be worth extracting a small private helper like tryConsume(ConcurrentMap<String, TbRateLimits> map, String clientConfig, String clientId) and having the three call sites supply their own map/config? The surrounding logic differs (logging, the QoS-0 gate), so only the two-line core would move, but it keeps the bucket-creation contract in one place.

[dmytro-landiak]

The sibling flags in this interface are isDevicePersistedMsgsLimitEnabled() and isTotalMsgsLimitEnabled(), but this one has an extra Rate: isApplicationPersistedMsgsRateLimitEnabled(). Minor, but for a set of methods a caller finds via autocomplete the odd-one-out is a small papercut — isApplicationPersistedMsgsLimitEnabled() would line up with the others.

[dmytro-landiak]

These four throttle tests use public visibility and given/when/then names, while the rest of this JUnit 5 file uses package-private methods and method_scenario_result names (e.g. processMainPack_retryAll_whenClientNeverAcks_...). Matching the surrounding convention keeps the file consistent for the next reader.