[Python] QoS 2 messages reach the broker but don't get forwarded

I've been struggling for a week now with an MQTT issue that seems both subtle and difficult to trace.

We have ~70 embedded units sending sensor data. Each unit uses the following message flow per session:

• Sends a "start" message to the backend, QoS 2
• Backend replies with "ack" to the unit, QoS 2
• Unit streams sensor data as multiple QoS 0 messages
• Backend eventually sends a "stop" to the unit, QoS 2

All of this is brokered through Mosquitto. The entire flow works as expected during the first round. But in the second round (or later), some units' "start" messages never reach the backend, even though:

• 
  The unit sends the "start" message (confirmed via Wireshark)
  
  
• 
  The broker receives it (seen in Mosquitto logs and wireshark)
  
  
• 
  The backend is subscribed to the correct topic (also confirmed via logs and client state)
  

The backend never receives the message.

Oddly enough, this doesn’t happen consistently. It seems random which units are affected.

Here I ask myself, Why would Mosquitto receive a QoS 2 message and not forward it to an online, subscribed client? What are possible reasons for this behavior, especially when:

• The topic and payload are the same as before
• Message IDs are different
• Clients are still connected and subscribed

Any insight or tools to diagnose this further would be hugely appreciated!

Here are the wireshark screenshots:

For a unit that failed:

Thanks in advance

My mqtt client is initialized like this:

def __init_client_tx(self) -> bool:

is_initialized = False
try:
client_id = self.cfg_mqtt.get("client_id", "")
if not isinstance(client_id, str):
raise TypeError("client_id must be a string")

api_version = getattr(paho.CallbackAPIVersion, "VERSION2", None)

valid_protocols = {paho.MQTTv31, paho.MQTTv311, paho.MQTTv5}
protocol = self.cfg_mqtt.get("protocol", paho.MQTTv5)
if protocol not in valid_protocols:
raise ValueError(f"Invalid protocol: {protocol}")

self.__client_tx = paho.Client(
callback_api_version=api_version,
client_id=client_id + "_tx",
protocol=protocol,
)
is_initialized = True

except (KeyError, AttributeError, TypeError, ValueError, Exception) as e:
pass

return is_initialized

def __init_client_rx(self) -> bool:

is_initialized = False
try:
client_id = self.cfg_mqtt.get("client_id", "")
if not isinstance(client_id, str):
raise TypeError("client_id must be a string")

api_version = getattr(paho.CallbackAPIVersion, "VERSION2", None)

valid_protocols = {paho.MQTTv31, paho.MQTTv311, paho.MQTTv5}
protocol = self.cfg_mqtt.get("protocol", paho.MQTTv5)
if protocol not in valid_protocols:
raise ValueError(f"Invalid protocol: {protocol}")

self.__client_rx = paho.Client(
callback_api_version=api_version,
client_id=client_id + "_rx",
protocol=protocol,
)
is_initialized = True

except (KeyError, AttributeError, TypeError, ValueError, Exception) as e:
pass

return is_initialized


And I connect to mqtt broker like this:

def __connect_tx(self) -> MQTTErrorCode:
connect_res: MQTTErrorCode = MQTTErrorCode.MQTT_ERR_UNKNOWN
try:
self.__client_tx.username_pw_set(
username=os.environ[self.cfg_mqtt["userAccess"]],
password=os.environ[self.cfg_mqtt["passwordAccess"]],
)

# Assign MQTT callbacks
self.__client_tx.on_connect = self.on_connect_tx
self.__client_tx.on_message = self.on_message_tx
self.__client_tx.on_disconnect = self.on_disconnect_tx

# Connect to the appropriate server
valid_envs = {"LOCAL", "DEVELOPMENT", "PREPRODUCTION", "PRODUCTION"}
if self.env not in valid_envs:
sys.exit("ERROR: Invalid environment! Choose from LOCAL, DEVELOPMENT, PREPRODUCTION, or PRODUCTION.")
else:
mqtt_host = self.cfg_mqtt["address"][self.env.lower()]
mqtt_port = self.cfg_mqtt["port"]

# Create a properties object for a specific packet type (e.g., CONNECT)
connect_properties = paho.Properties(paho.PacketTypes.CONNECT)
# Set MQTT 5 properties
connect_properties.SessionExpiryInterval = 600 # Session expires after 10 min
connect_properties.ReceiveMaximum = 5 # Limit max QoS1/QoS2 messages received
connect_properties.MaximumPacketSize = 65536 # Max packet size in bytes
connect_properties.TopicAliasMaximum = 10 # Allow up to 10 topic aliases
connect_properties.RequestProblemInformation = 1 # Request problem info from broker
connect_res = self.__client_tx.connect(
host=mqtt_host,
port=mqtt_port,
clean_start=paho.MQTT_CLEAN_START_FIRST_ONLY,
properties=connect_properties,
)
except Exception as e:
pass
return connect_res

def __connect_rx(self) -> MQTTErrorCode:
connect_res: MQTTErrorCode = MQTTErrorCode.MQTT_ERR_UNKNOWN
try:
self.__client_rx.username_pw_set(
username=os.environ[self.cfg_mqtt["userAccess"]],
password=os.environ[self.cfg_mqtt["passwordAccess"]],
)

# Assign MQTT callbacks
self.__client_rx.on_connect = self.on_connect_rx
self.__client_rx.on_message = self.on_message_rx
self.__client_rx.on_disconnect = self.on_disconnect_rx

# Connect to the appropriate server
valid_envs = {"LOCAL", "DEVELOPMENT", "PREPRODUCTION", "PRODUCTION"}
if self.env not in valid_envs:
sys.exit("ERROR: Invalid environment! Choose from LOCAL DEVELOPMENT, PREPRODUCTION, or PRODUCTION.")
else:
mqtt_host = self.cfg_mqtt["address"][self.env.lower()]
mqtt_port = self.cfg_mqtt["port"]

# Create a properties object for a specific packet type (e.g., CONNECT)
connect_properties = paho.Properties(paho.PacketTypes.CONNECT)
# Set MQTT 5 properties
connect_properties.SessionExpiryInterval = 600 # Session expires after 10 min
connect_properties.ReceiveMaximum = 5 # Limit max QoS1/QoS2 messages received
connect_properties.MaximumPacketSize = 65536 # Max packet size in bytes
connect_properties.TopicAliasMaximum = 10 # Allow up to 10 topic aliases
connect_properties.RequestProblemInformation = 1 # Request problem info from broker
connect_res = self.__client_rx.connect(
host=mqtt_host,
port=mqtt_port,
clean_start=paho.MQTT_CLEAN_START_FIRST_ONLY,
properties=connect_properties,
)
except Exception as e:
pass
return connect_res


I have tried:

• Changing broker and client's configuration
• Decreasing the qos 0 messages sending speed
• Logging whenever the client enters in on_unsubscribe and on_disconnect methods (it does not enter)
• Logging on_message incoming messages (the expected message doesn't reach)
• Checked on wireshark and indeed, the message reaches the broker but the broker does not forward the message to the expected client.

I expect the broker forwarding this message to the client.

Continue reading...