Persistent APPLICATION client

This page explains how TBMQ routes and delivers messages for persistent APPLICATION clients, including the per-client Kafka topic design, consumer thread model, pack processing, and shared subscription support.

APPLICATION clients are designed for backend services that consume high-rate topic streams and require guaranteed delivery of every message, including those published while the client was offline. Unlike persistent DEVICE clients, which share a single Kafka topic and use Redis for offline storage, each APPLICATION client gets its own dedicated Kafka topic. This design isolates processing per client, enabling independent scaling and high per-client throughput.

When a publisher sends a PUBLISH message, it is first written to the tbmq.msg.all Kafka topic. A Kafka consumer thread reads the message, queries the Subscription Trie for matching subscribers, and routes messages for APPLICATION clients to their dedicated topics:

tbmq.msg.all  ──▶  Message dispatcher  ──▶  tbmq.msg.app.$CLIENT_ID

The dedicated topic is created automatically when the client authenticates as an APPLICATION type and connects with a persistent session — the broker creates it as the first step of starting that client’s message processing, and later deletes both the topic and its consumer group when the client stops being a persistent APPLICATION client. If the client ID contains characters outside the alphanumeric range, its SHA-256 hash is used instead:

tbmq.msg.app.$CLIENT_ID        ← alphanumeric client IDs
tbmq.msg.app.$CLIENT_ID_HASH   ← client IDs with special characters

The hash-based naming is controlled by TB_APP_PERSISTED_MSG_CLIENT_ID_VALIDATION (enabled by default). The check is strictly alphanumeric (^[a-zA-Z0-9]+$), so an ID containing - or _ is hashed too, even though Kafka itself would accept those characters — which is why topics for such clients look like a hash rather than the client ID. Disabling the validation makes TBMQ pass the raw client ID to Kafka, and Kafka rejects any name outside [a-zA-Z0-9._-], so the topic is never created and the client cannot receive persisted messages.

Each per-client topic has exactly one partition. TBMQ overrides any partitions value configured in TB_KAFKA_APP_PERSISTED_MSG_TOPIC_PROPERTIES, because a single partition is what makes the ordering guarantee below hold. Scale comes from having many topics, not many partitions per topic.

Offline retention is therefore Kafka retention: retention.ms 7 days and retention.bytes 1 GB per client topic by default. A client that stays offline longer than that loses the oldest messages — plan retention around the longest outage you intend to survive.

A dedicated Kafka consumer thread is assigned to each persistent APPLICATION client when it connects. This thread:

1. Continuously polls tbmq.msg.app.$CLIENT_ID for new messages (TB_APP_PERSISTED_MSG_POLL_INTERVAL, 100 ms).
2. Processes messages in packs (batches) — delivering each pack and waiting for acknowledgements before advancing the consumer offset. See Pack processing and acknowledgement for details.
3. Maintains delivery order — Kafka’s single-partition consumer guarantee preserves message sequence.

Threads come from two cached pools — application-persisted-msg-consumers for the per-client main consumers and application-shared-subs-msg-consumers for shared subscription consumers — so threads are reused across connect/disconnect cycles instead of being created per session.

Each client’s consumer joins its own consumer group, application-persisted-msg-consumer-group-$CLIENT_ID, which is what carries the delivery position between sessions. On the very first start, when the group has no committed offset yet, TBMQ commits the topic’s end offset — a brand-new persistent session starts from “now” rather than replaying whatever the topic still retains.

This per-client thread model ensures that a slow or high-volume client does not impact other clients. In cluster mode, the dedicated consumer is created on the node where the client connects — no inter-node message forwarding is required, unlike the DEVICE client flow.

The consumer thread delivers messages in packs — batches polled from Kafka in a single call. The maximum number of records per poll is controlled by max.poll.records, set through TB_KAFKA_APP_PERSISTED_MSG_ADDITIONAL_CONSUMER_CONFIG (default 200 for a client’s own topic; TB_KAFKA_APP_PERSISTED_MSG_SHARED_ADDITIONAL_CONSUMER_CONFIG raises it to 500 for shared subscription topics). Each pack goes through a deliver-wait-commit cycle:

1. Poll — the consumer fetches the next batch of messages from the client’s Kafka topic. Any PUBREL messages left pending by the previous pack (or restored from the previous session) are merged in ahead of the new PUBLISH messages, so an interrupted QoS 2 exchange finishes before new data flows.
2. Deliver — messages are sent to the client over the MQTT connection. Expired messages are dropped here rather than delivered (see Message expiry), so a pack can be smaller than the polled batch.
3. Wait — the broker waits for the client to acknowledge every message in the pack. The wait timeout is controlled by TB_APP_PERSISTED_MSG_PACK_PROCESSING_TIMEOUT (default 20 seconds).
4. Commit or retry — once all messages are acknowledged, the Kafka consumer offset is committed and the next pack is polled. If some messages remain unacknowledged after the timeout, the acknowledgement strategy decides what happens next.

Because the offset advances only for a fully acknowledged pack, a crash mid-pack costs at most a redelivery, never a loss. It also means a client that stops acknowledging blocks its own stream and nothing else.

The acknowledgement the broker expects depends on the message’s QoS level:

• QoS 0 — nothing to acknowledge, so the message never enters the pack’s pending set and cannot hold up the commit. This only arises on a shared subscription: a client’s own topic always carries QoS 1 or 2, since a message is routed there only when both the publish and the subscription used QoS > 0.
• QoS 1 — the client sends a PUBACK. The message is removed from the pending map.
• QoS 2 — a four-step handshake: the broker delivers the PUBLISH, the client sends PUBREC, the broker responds with PUBREL, and the client completes the exchange with PUBCOMP. Only after PUBCOMP is the message fully acknowledged.
• QoS 2, refused — a PUBREC carrying an MQTT 5 error reason code (0x80 or higher) means the client is rejecting the message. TBMQ settles the packet without sending PUBREL, so the pack can complete and commit instead of waiting out the timeout on a message the client will never accept.

The acknowledgement strategy (TB_APP_PERSISTED_MSG_ACK_STRATEGY_TYPE) determines behavior when the pack processing timeout expires with unacknowledged messages:

• RETRY_ALL (default) — resubmits unacknowledged messages with the MQTT DUP flag set. The maximum number of retries is controlled by TB_APP_PERSISTED_MSG_ACK_STRATEGY_RETRIES (default 3; set to 0 for unlimited retries). When all retries are exhausted, the pack is committed to prevent indefinite blocking.
• SKIP_ALL — discards unacknowledged messages and commits the Kafka offset immediately.

Only the still-unacknowledged messages are resubmitted; already-acknowledged ones are dropped from the retry set. PUBREL messages are re-sent as they are — the DUP flag applies to PUBLISH only.

Whenever a pack is committed with messages still unacknowledged — the retry budget exhausted, or SKIP_ALL — those messages are counted as dropped and a warning is logged naming the skipped offsets. Persistent gaps in a client’s stream are visible there and in the broker’s dropped-message statistics. Setting the retry count to 0 (unlimited) trades that data loss for a stream that stalls until the client acknowledges.

TBMQ applies backpressure when a client reads slower than the message arrival rate — for a detailed treatment see Backpressure. The mechanism uses Netty’s write buffer watermarks:

• When the client’s TCP write buffer exceeds the high watermark (NETTY_WRITE_BUFFER_HIGH_WATER_MARK, default 1280 KB), all Kafka consumers for that client — both the main consumer and any shared subscription consumers — are paused.
• When the buffer drops below the low watermark (NETTY_WRITE_BUFFER_LOW_WATER_MARK, default 640 KB), consumers resume polling.

This prevents unbounded memory growth on the broker side when a client cannot keep up with the incoming message rate.

When a persistent APPLICATION client disconnects, the broker persists all unacknowledged message packet IDs and their Kafka offsets to PostgreSQL — the application_session_ctx table, one row per client, with two JSON columns: publish_msg_infos for PUBLISH messages awaiting PUBACK (QoS 1) or PUBREC (QoS 2), and pubrel_msg_infos for PUBREL messages awaiting PUBCOMP (QoS 2). Each entry is just a (packetId, offset) pair — the payloads are not copied, because Kafka still holds them at those offsets.

On reconnect:

1. The saved context is loaded from the database.
2. The client’s MQTT packet ID sequence is restored from the last persisted packet ID, preventing ID collisions with in-flight messages.
3. Previously unacknowledged messages are re-delivered with the same packet IDs and the DUP flag set. The match is made by Kafka offset: as the consumer re-reads from the committed offset, any record whose offset appears in the loaded context is delivered with its original packet ID instead of a fresh one.
4. After the recovered messages are processed, the consumer continues polling new messages from the committed Kafka offset.

The row is deleted when the session is cleared, so a clean start connect begins with no recovery state. Combined with Kafka’s committed consumer offset, this guarantees that no messages are lost between sessions.

Messages with an MQTT 5.0 message expiry interval are checked before delivery. If the interval has elapsed, the message is skipped — it is not delivered and does not consume a packet ID. For non-expired messages, the remaining expiry interval is recalculated and included in the delivered message properties, so the client receives an accurate remaining TTL.

TBMQ supports two delivery modes for APPLICATION clients, controlled by MQTT_APP_MSG_WRITE_AND_FLUSH:

• Write-and-flush (true) — flushes the Netty channel after each message. Provides the lowest latency for individual message delivery.
• Buffered (false, default for APPLICATION clients) — buffers messages and flushes every N messages, where N is controlled by MQTT_APP_BUFFERED_MSG_COUNT (default 10). Provides higher throughput for batch-oriented consumers by reducing the number of system calls. Whatever remains below the threshold at the end of a pack is flushed explicitly, so buffering never delays the tail of a batch.

Buffered is the default here precisely because an APPLICATION client is expected to consume a high-rate stream, which is the opposite default from persistent DEVICE clients. See Message delivery strategies for the full set of options.

The dedicated-topic-per-client architecture makes shared subscriptions efficient for APPLICATION clients. When multiple APPLICATION clients share a subscription, each client gets its own consumer in the same Kafka consumer group, and Kafka handles partition-based load balancing natively.

Unlike a client’s own topic, a shared subscription topic must be provisioned in advance: an Application Shared Subscription entity has to exist before a persistent APPLICATION client may subscribe, and its partition count is fixed at creation. That count is the ceiling on parallelism — Kafka gives each partition to exactly one consumer in the group, so with 3 partitions a fourth subscriber sits idle until another leaves. Choose it from the expected subscriber count and leave room to grow; see Application shared subscriptions. Records are produced to these topics without a key, so they spread across partitions and per-topic ordering is not preserved across the group — the ordering guarantee applies to a client’s own topic, not to a shared subscription.

Each shared subscription receives its own independent Kafka consumer, processing thread, and pack processing context — fully isolated from the main consumer and from other shared subscriptions on the same client. This means a client can have one main consumer for its dedicated topic and multiple shared subscription consumers running concurrently.

The shared subscription Kafka topic follows the naming convention:

Where # maps to mlw (multi-level wildcard) and + maps to slw (single-level wildcard). The accepted set here is ^[a-zA-Z0-9/+#]+$ — wider than the client-ID check, because separators and wildcards are translated rather than rejected. A filter containing anything else (-, _, ., …) is replaced wholesale by the SHA-256 hash of the filter. This behavior is controlled by TB_APP_PERSISTED_MSG_SHARED_TOPIC_VALIDATION (enabled by default).

All APPLICATION clients sharing the same subscription join the same Kafka consumer group: application-shared-msg-consumer-group-{shareName}-{sharedAppTopic}. Kafka distributes topic partitions among the consumers in the group, providing native load balancing without any broker-level coordination.

The effective QoS of a delivered message is always the minimum of the publisher’s QoS and the subscriber’s requested QoS. The difference is when it is applied: for a client’s own topic the routed copy already carries the resolved QoS, whereas a shared subscription topic is shared by subscribers that may have requested different levels, so the downgrade is applied per subscriber at delivery time.

The properties for APPLICATION client Kafka topics are controlled by:

TB_KAFKA_APP_PERSISTED_MSG_TOPIC_PROPERTIES        # per-client topics
TB_KAFKA_APP_PERSISTED_MSG_SHARED_TOPIC_PROPERTIES # shared subscription topics

Both default to retention.ms:604800000 (7 days), retention.bytes:1048576000 (1 GB), segment.bytes:26214400, and replication.factor:1 — raise the replication factor on a real Kafka cluster, otherwise the loss of one Kafka broker takes the messages with it. partitions is not configurable for per-client topics (always 1, see Message routing); for shared subscription topics it comes from the Application Shared Subscription entity. See the full configuration reference for details.

A client’s dedicated topic outlives a single session on purpose — that is what makes offline delivery possible — so it is removed only when the client is no longer a persistent APPLICATION client, for example after its credentials are switched to the DEVICE type. Because deleting a Kafka topic is disruptive and cannot be undone, TBMQ does not do it inline: it publishes an event to tbmq.sys.app.removed, and a scheduled job (TB_APPLICATION_REMOVED_EVENT_PROCESSING_CRON, daily at 03:00 UTC by default) drains those events and requests the removal. Any messages still persisted for that client are cleared at the moment of the change.

APPLICATION clients behave identically in standalone and cluster modes. Because each client’s Kafka consumer is created on the node where the client connects, message processing is local — no downlink Kafka topic is needed for inter-node forwarding (unlike DEVICE clients). This design enables horizontal scaling without coordination overhead: adding broker nodes distributes APPLICATION client connections automatically.

For the opposite trade-off — a shared Kafka topic plus Redis, sized for a very large number of intermittently connected clients — see persistent DEVICE clients.