Durable session max 100 messages per batch

[Saddamus]

Is there any reason why in opensource version client can not pull more than 100 messages per batch ? I have changed max inflight window, messagre query batch size, but whatever i do, telegraf can not pull more than 100 messages per second (exaclty 100, not 99, neither 101) and it is not telegraf limitation, because when using normal persistant session, it can pull plenty times more messages. Is there any fixed limit on durable session for opensource emqx ?

[ieQu1]

Hello,

There are no artificial limitations on the throughput for open-source edition. These symptoms can happen if the client doesn't pipeline PUBACKs. Can you collect a wireshark trace of the telegraf MQTT session?

[Saddamus]

Thanx for your response. Im more than sure its not question of telegraf as without durable session, same stresstest env, same qos it can pull thousands of messages per second. If i enabled debug log on emqx i can see lines like stage-broker | 2025-02-07T09:49:34.185254+00:00 [debug] msg: cluster_rpc_apply_result, result: {ok,#{config => #{file => #{default => #{time_offset => "system",path => "${EMQX_LOG_DIR}/emqx.log",level => warning,chars_limit => unlimited,single_line => true,formatter => text,drop_mode_qlen => 3000,flush_qlen => 8000,sync_mode_qlen => 100,enable => false,with_mfa => false,timestamp_format => auto,payload_encode => text,burst_limit => #{enable => true,max_count => 10000,window_time => 1000},max_depth => 100,overload_kill => #{enable => true,mem_size => 31457280,qlen => 20000,restart_after => 5000},supervisor_reports => error,rotation_count => 10,rotation_size => 52428800}},console => #{time_offset => "system",level => debug,chars_limit => unlimited,single_line => true,formatter => text,drop_mode_qlen => 3000,flush_qlen => 8000,sync_mode_qlen => 100,enable => true,with_mfa => false,timestamp_format => auto,payload_encode => text,burst_limit => #{enable => true,max_count => 10000,window_time => 1000},max_depth => 100,overload_kill => #{enable => true,mem_size => 31457280,qlen => 20000,restart_after => 5000},supervisor_reports => error},throttling => #{msgs => [authentication_failure,authorization_permission_denied,cannot_publish_to_topic_due_to_not_authorized,cannot_publish_to_topic_due_to_quota_exceeded,connection_rejected_due_to_license_limit_reached,data_bridge_buffer_overflow,dropped_msg_due_to_mqueue_is_full,external_broker_crashed,failed_to_retain_message,handle_resource_metrics_failed,socket_receive_paused_by_rate_limit,unrecoverable_resource_error],time_window => 60}},post_config_update => #{kind => initiate},raw_config => #{<<"console">> => #{<<"enable">> => true,<<"formatter">> => <<"text">>,<<"level">> => <<"debug">>,<<"payload_encode">> => <<"text">>,<<"time_offset">> => <<"system">>,<<"timestamp_format">> => <<"auto">>},<<"file">> => #{<<"default">> => #{<<"enable">> => false,<<"formatter">> => <<"text">>,<<"level">> => <<"warning">>,<<"path">> => <<"${EMQX_LOG_DIR}/emqx.log">>,<<"payload_encode">> => <<"text">>,<<"rotation_count">> => 10,<<"rotation_size">> => <<"50MB">>,<<"time_offset">> => <<"system">>,<<"timestamp_format">> => <<"auto">>}},<<"throttling">> => #{<<"time_window">> => <<"1m">>}}}}, kind: initiate, tnx_id: 1, entrypoint: <<"emqx:update_config/3">> You can see there sync_mode_qlen => 100 and this is the most likely what limits telegraf to 100 messages per batch. No idea where this limitation comes from, because on settings i do not have anything that can seem to be related.

[Saddamus]

emqx-qos1-durable.zip

[ieQu1]

Thanks for the trace. It confirms my initial hypothesis. I see some rather strange behavior on the client side:

1. In the highlighted fragment at t=0.004293 EMQX sends a batch of messages to the client with the first packet id=53202 and last packet id=53300 (98 packets in total)
2. The client sends PUBREC only a second later, at t=1.000064.

That pause interrupts the pipelining in the durable session, which processes data in batches.

I assume client's ReceiveMaximum is 100.

How to improve the throughput:

1. I am not familiar with telegraf, but it definitely has some batching of its own, that causes the 1s delay. Try to adjust the settings to eliminate or lower this interval.
2. Increase receive maximum on the client (if it's using MQTTv5), and/or broker: https://docs.emqx.com/en/emqx/latest/design/inflight-window-and-message-queue.html#inflight-window-and-receive-maximum
3. Publishing data in parallel (via multiple topics and/or publishers) will also help.

[Saddamus]

I'm not sure if i fully understand your findings, but this does not prevent me from supposition you are wrong or at least partially wrong :) Lets focus at point 1 You already see by first and last message ID that emqx is letting telegraf fetch exaclty 98 messages, despite the input is with thousands per second. Telegraf is getting these messages, saving to DB and comes back 1 second later for new ones and then it will again receive exactly 100 messages or very close to 100.
The limitation is not coming from telegraf, becuse in persisant session (but not durable) it can pull even 5 K messages per second with no issue. I would rather ask why on your highlighted screen, emqx is giving ~100 messages

Then if we go further and assume that statement regarding 1 secod delay interrupts pipelining in durable session is correct, why telegraf can keep picking up 100 or very close to 100 messages on every session / every second? This doesnt make sense, why the number is so close to 100 and so stable for each batch ?

Teegraf supports mqtt v3

[ieQu1]

but this does not prevent me from supposition you are wrong or at least partially wrong :)

I'm the one who designed the durable sessions' flow control mechanism, and I know exactly what's going on in your case 🤷

despite the input is with thousands per second.

Input volume is not important in this case. When a durable session subscribes to a topic, this topic is marked as durable, and messages sent to it are saved in a builtin DB. The subscriber reads the messages from the DB in batches at its own pace. Producer and consumer are totally insulated from one another:

publisher -> [DB]

...

             [DB] -> subscriber

If the subscriber doesn't consume messages fast enough (e.g. due to waiting 1s before sending PUBREC/PUBCOMP), it will lag behind, until eventually the message retention policy removes the old messages, and the consumer skips ahead to the oldest message still available.

This doesnt make sense, why the number is so close to 100 and so stable for each batch ?

Because the maximum number of inflight (sent, but not acked) packets is governed by combination of max_inflight parameter on the broker, and ReceiveMaximum property set by the client in the CONNECT packet (MQTTv5-only), whichever is less. It's a constant.

Additionally, Message Query Batch Size configuration parameter can play its role as well. It's also a constant. By the way, you can try to increase it as well.

The difference in performance of durable and regular session in presence of the 1s delay is explained by a fundamental difference in their implementation. Memory session keeps all runtime data in RAM and processes messages individually. Durable session reads messages from disk in batches. Memory session disappears during restart of the broker, durable session backs up its metadata on disk and thus survives restart of the broker.

Since bandwidth of disk is greatly limited compared to the bandwidth of RAM, we had to derive some strategies for minimizing the volume of metadata per session. One optimization is storing the metadata only for the last batch per stream. (Sort of like storing one offset per Kafka partition.) This means the client should ack all messages in the batch before receiving the next batch from the same stream. In your case there's only one publisher and only one topic, meaning there's only one stream. Which gets blocked by delayed acks.

TL;DR:

1. 1s delay is the problem
2. You can try to mitigate it by raising max_inflight and Message Query Batch Size
3. But the real solution is removing the 1s delay on the client side.

[Saddamus]

OK, thanks to your response, i have sorted it out by decreasing batch size on output, so that messages are dumped to output i smaller portions and now telegraf can pickup much more. I have one offtop question. Directory /data/messages with durable session is constantly growing and not sure what settings parameter is purposed to clean up messages database. Retention period is rather related to message expiration, but its not described if expired messages are being deleted from disk. There is also max_queue, but it doesnt bring influence on growing storage usage. Can you help me to understand how messages are cleaned ? I thought that qos2 message should be deleted from storage once its received, but its not the case.

[ieQu1]

You can adjust durable_sessions.message_retention_period: https://docs.emqx.com/en/emqx/latest/durability/durability_introduction.html#generation

This value governs frequency of rotating the generations. When old generation is deleted, it's gone from the file system.

Warning: unit of this config parameter is in milliseconds, so make sure to specify the unit, like 1d or 1h.

P.S.

I thought that qos2 message should be deleted from storage once its received, but its not the case.

1. The same message can be sent to multiple durable clients. EMQX will save only one copy of this message. As such, there isn't a trivial way to track ownership of the messages: there could be thousands of clients that need one message. This could make garbage collection very expensive.
2. Another problem lies in how RocksDB works internally. Deletion of records is implemented via tombstones: by deleting individual messages, we'd only create more data.

These problems aren't impossible to solve, but for time being we settled on a simple and efficient time-based retention policy.