Msg delivery strategies

When a client subscribes to a topic and a matching message is published, TBMQ delivers the message through its underlying networking layer — Netty. Netty provides two ways to send data over a channel:

• writeAndFlush() — sends the message and immediately flushes the channel.
• write() — writes to the channel’s buffer without sending. Data is held until an explicit flush() call.

TBMQ uses this to offer two configurable delivery strategies:

• Write and flush for each message
• Buffer messages and flush periodically or by count

The strategy is chosen independently for each of the three outbound delivery paths:

TBMQ calls writeAndFlush() for every outgoing message.

When to use: Low-throughput environments where minimal latency matters more than throughput efficiency.

Trade-off: High CPU and I/O overhead under heavy load due to excessive flush operations.

write() queues the message without flushing. TBMQ triggers a flush when:

• The buffer reaches the configured message count (buffered-msg-count).
• More than idle-session-flush-timeout-ms has passed since the last flush — direct and persistent DEVICE paths only.
• The current Kafka batch is fully processed — persistent APPLICATION path only.

When to use: High-throughput or bursty workloads where I/O efficiency and scalability are priorities.

Trade-off: May introduce delivery delays in low-throughput scenarios where messages arrive infrequently.

These two paths share one buffering implementation, each with its own pair of settings:

mqtt:
  # Enables or disables immediate flush of each message published to non-persistent subscribers.
  # When disabled, messages are buffered and flushed periodically.
  write-and-flush: "${MQTT_MSG_WRITE_AND_FLUSH:true}"
  # Number of messages buffered in the channel before the flush is made.
  # Used when MQTT_MSG_WRITE_AND_FLUSH = false.
  buffered-msg-count: "${MQTT_BUFFERED_MSG_COUNT:5}"
  persistent-session:
    device:
      persisted-messages:
        # Enables or disables immediate flush of each message published to persistent DEVICE subscribers.
        write-and-flush: "${MQTT_PERSISTENT_MSG_WRITE_AND_FLUSH:true}"
        # Number of messages buffered in the channel before the flush is made.
        # Used when MQTT_PERSISTENT_MSG_WRITE_AND_FLUSH = false.
        buffered-msg-count: "${MQTT_PERSISTENT_BUFFERED_MSG_COUNT:5}"

When either of these two flags is false, TBMQ builds a cache of active session states and starts a background flusher. When both are true — the default — neither the cache nor the scheduler is created at all, so buffered delivery adds no overhead. The parameters below control the cache and the flusher:

mqtt:
  buffered-delivery:
    # Maximum number of session entries stored in the flush state cache.
    # When the cache exceeds this size, the least recently used sessions are evicted
    # and their pending message buffers are flushed automatically.
    session-cache-max-size: "${MQTT_BUFFERED_CACHE_MAX_SIZE:10000}"
    # Expiry time for an inactive session entry in the flush cache (in ms).
    # A session is considered inactive if it receives no new messages during this period.
    # Upon expiration, the session is evicted from the cache and its buffer is flushed.
    # Defaults to 5 minutes.
    session-cache-expiration-ms: "${MQTT_BUFFERED_CACHE_EXPIRY_MS:300000}"
    # Interval at which the scheduler checks all sessions in the cache for potential flushing (in ms).
    scheduler-execution-interval-ms: "${MQTT_BUFFERED_SCHEDULER_INTERVAL_MS:100}"
    # Maximum time a session can remain idle before its message buffer is automatically flushed (in ms).
    idle-session-flush-timeout-ms: "${MQTT_BUFFERED_IDLE_FLUSH_MS:200}"

Buffered delivery on these paths works as follows:

1. Session buffer creation — TBMQ stores a flush-state entry in the cache, keyed by session ID, holding the buffered message count, the last flush timestamp, and the client’s Netty channel context.
2. Write without flush — each outgoing PUBLISH is written with channel.write() and increments the buffer count. Only PUBLISH packets are buffered on these two paths: PUBREL, the second phase of a QoS 2 delivery, and retained messages sent in response to a SUBSCRIBE are always flushed immediately.
3. Count-based flush — a flush is triggered once the count reaches buffered-msg-count, after which the counter resets.
4. Time-based flush — the background scheduler runs every scheduler-execution-interval-ms and flushes any session whose buffer is non-empty and whose last flush was more than idle-session-flush-timeout-ms ago.
5. Eviction flush — a session evicted from the cache, by size pressure or after session-cache-expiration-ms of inactivity, has its buffer flushed on the way out.
6. Shutdown handling — on service shutdown, TBMQ flushes all buffered sessions to prevent message loss.

If sessions are evicted because the cache is full, the broker logs an INFO message — Client session was evicted due to cache size limit — followed by the configured limit. Seeing it often means MQTT_BUFFERED_CACHE_MAX_SIZE is below your concurrent subscriber count and should be raised.

mqtt:
  persistent-session:
    app:
      persisted-messages:
        # Enables or disables immediate flush of each message published to persistent APPLICATION subscribers.
        write-and-flush: "${MQTT_APP_MSG_WRITE_AND_FLUSH:false}"
        # Number of messages buffered in the channel before the flush is made.
        # Used when MQTT_APP_MSG_WRITE_AND_FLUSH = false.
        buffered-msg-count: "${MQTT_APP_BUFFERED_MSG_COUNT:10}"

This path does not use the shared cache or the scheduler. Each APPLICATION client is processed in a dedicated consumer thread that polls a batch of messages from the client’s own Kafka topic, so flushing can be controlled per client inside the batch:

1. Each message in the batch — both PUBLISH and PUBREL packets — is written to the Netty channel without an immediate flush.
2. A flush is triggered every buffered-msg-count messages (default: 10). The counter is local to the batch and restarts at zero on every poll, so a value larger than the batch — polls are capped by the consumer’s max.poll.records, 200 by default for a client’s own topic — never fires a flush of its own.
3. When the batch finishes, any remaining unflushed messages are flushed explicitly.

Because a flush always happens at the end of every batch, there is no idle-timeout mechanism here: a message is never left buffered waiting for more traffic. This is why buffering is the default for APPLICATION clients but not for the other two paths.

Use write-and-flush = true when:

• Low latency is the priority over throughput.
• Your system experiences low to moderate message rates.
• Clients expect immediate delivery, like real-time dashboards or alerts.
• Simplicity and predictability matter more than raw performance.

Use write-and-flush = false (buffered delivery) when:

• You expect high-throughput workloads with frequent publications.
• Minimizing system call overhead and I/O pressure is important.
• Clients can tolerate slight delivery delays in exchange for improved efficiency.
• You want to scale to thousands of clients without saturating the CPU or network.

Tuning tips:

• Start with buffered-msg-count between 5 and 10 and adjust based on profiling.
• On the direct and persistent DEVICE paths, tune idle-session-flush-timeout-ms to balance delay against timely delivery — it is the ceiling on how long a message can sit buffered.
• Watch the broker log for flush-state cache evictions caused by the size limit, and raise MQTT_BUFFERED_CACHE_MAX_SIZE if they are frequent.
• If messages are frequently delayed in low-throughput setups, enable immediate flushing.