I am trying to transfer data from a sensor to our backend over MQTT, and it should not lose data. Setup:
Self-hosted EMQX cluster, 3 nodes. Max queue size is the default 1000 messages.
One producer, let’s say it is just a counter publishing messages 1, 2, 3 etc, with an interval of 10 per second. The topic has QoS 1.
One consumer, that subscribes to the same topic. It’s a persistent session, with a long expiry interval (let’s say 3600 seconds).
Procedure:
The producer is started, it creates the messages with increasing integers.
The consumer is started for the first time, it picks up from the most recent integer (10).
The consumer (or its connection) is interrupted, it does not consume data after 15.
The producer keeps on publishing integers.
The consumer is restarted much later when 2000 is published. It receives integers 1000 to 2000 immediately, but lacks 15 to 1000 because they were dropped out of the queue.
One exception are messages that were ‘in flight’: already sent but not yet acknowledged. These are the last few, so depending on the timing, the consumer may also receive 15, 16 and a few more.
Is there a way that the producer can be made aware that the broker is losing messages?
Persistent sessions in MQTT are subscriber-centered. When a message is published to some topic, there is no difference, from the publisher’s perspective, if the topic has subscribers with or without persistent session, or if there are no subscribers at all. MQTT protocol doesn’t have any means of letting the publishers know what happens to the published messages on the subscriber side.
(Same as in Kafka, for example)
Perhaps the solution that you’re looking for, is creating a different topic where the subscriber can report progress its progress of consuming the data, that the publisher can monitor?
Thanks @dmif! That is indeed an interesting idea, to have the subscriber post its status (possibly using LWT) to a topic to notify publishers when they should slow down.
I have a follow-up question on this.
We have implemented this and it works for our custom solution, but it doesn’t work with some off-the-shelf hardware we are using.
We would like to be able to let off-the-shelf hardware buffer messages locally once messages are no longer being consumed from the broker. Is it possible to set up a system where we can auto-reject all messages sent to a certain topic (with wildcard)?
i.e. Can we auto-reject all messages sent to sensors/weather/# once the consumer is no longer consuming?
If not, is there a different solution that could work?
