Merge branch 'main' into ct-requests-unixsocket

Author: cartertinney

azure-iot-device/azure/iot/device/common/mqtt_transport.py

@@ -140,6 +140,7 @@ def _create_mqtt_client(self):
         if self._websockets:
             logger.info("Creating client for connecting using MQTT over websockets")
             mqtt_client = mqtt.Client(
+                callback_api_version=mqtt.CallbackAPIVersion.VERSION1,
                 client_id=self._client_id,
                 clean_session=False,
                 protocol=mqtt.MQTTv311,
@@ -149,7 +150,10 @@ def _create_mqtt_client(self):
         else:
             logger.info("Creating client for connecting using MQTT over TCP")
             mqtt_client = mqtt.Client(
-                client_id=self._client_id, clean_session=False, protocol=mqtt.MQTTv311
+                callback_api_version=mqtt.CallbackAPIVersion.VERSION1,
+                client_id=self._client_id,
+                clean_session=False,
+                protocol=mqtt.MQTTv311,
             )
 
         if self._proxy_options:
@@ -436,7 +440,6 @@ def disconnect(self, clear_inflight=False):
         :raises: ConnectionDroppedError in unexpected cases.
         :raises: UnauthorizedError in unexpected cases.
         :raises: ConnectionFailedError in unexpected cases.
-        :raises: NoConnectionError if the client isn't actually connected.
         """
         logger.info("disconnecting MQTT client")
         try:
@@ -452,11 +455,22 @@ def disconnect(self, clear_inflight=False):
 
         logger.debug("_mqtt_client.disconnect returned rc={}".format(rc))
         if rc:
-            # This could result in ConnectionDroppedError or ProtocolClientError
-            # No matter what, we always raise here to give upper layers a chance to respond
-            # to this error.
-            err = _create_error_from_rc_code(rc)
-            raise err
+            # Special case: MQTT_ERR_NO_CONN (rc=4) during disconnect means the socket
+            # is already closed. In Paho 2.x, this can happen even after a successful
+            # disconnect because the on_disconnect callback fires (with rc=0) before
+            # disconnect() returns, and Paho's internal cleanup closes the socket.
+            # Since we wanted to disconnect and we're disconnected, treat this as success.
+            if rc == mqtt.MQTT_ERR_NO_CONN:
+                logger.debug(
+                    "disconnect returned MQTT_ERR_NO_CONN - socket already closed, treating as success"
+                )
+                # Still clear inflight operations since we're effectively disconnected
+                if clear_inflight:
+                    self._op_manager.cancel_all_operations()
+            else:
+                # This could result in ConnectionDroppedError or ProtocolClientError
+                err = _create_error_from_rc_code(rc)
+                raise err
         else:
             # Clear pending ops if instructed, but only if the disconnect was successful.
             # Technically the disconnect could still fail upon response, however that would then

setup.py

@@ -72,7 +72,7 @@
     ],
     install_requires=[
         "deprecation>=2.1.0,<3.0.0",
-        "paho-mqtt>=1.6.1,<2.0.0",
+        "paho-mqtt>=2.0.0,<3.0.0",
         "requests>=2.32.3,<3.0.0",
         "requests-unixsocket>=0.4.1",
         "janus",

tests/e2e/iothub_e2e/conftest.py

@@ -67,7 +67,7 @@ def random_reported_props():
 all_objects_can_leak = []
 one_object_can_leak = [
     "<class 'azure.iot.device.common.alarm.Alarm'>",
-    "<class 'paho.mqtt.client.WebsocketWrapper'>",
+    "<class 'paho.mqtt.client._WebsocketWrapper'>",
 ]
 
 

tests/unit/common/test_mqtt_transport.py

@@ -124,6 +124,12 @@
     },
 ]
 
+# For disconnect, MQTT_ERR_NO_CONN is treated as success (socket already closed)
+# so we exclude it from the error return codes for disconnect tests
+disconnect_operation_return_codes = [
+    x for x in operation_return_codes if x["rc"] != mqtt.MQTT_ERR_NO_CONN
+]
+
 
 @pytest.fixture
 def mock_mqtt_client(mocker, fake_paho_thread):
@@ -202,7 +208,10 @@ def test_instantiates_mqtt_client(self, mocker):
 
         assert mock_mqtt_client_constructor.call_count == 1
         assert mock_mqtt_client_constructor.call_args == mocker.call(
-            client_id=fake_device_id, clean_session=False, protocol=mqtt.MQTTv311
+            callback_api_version=mqtt.CallbackAPIVersion.VERSION1,
+            client_id=fake_device_id,
+            clean_session=False,
+            protocol=mqtt.MQTTv311,
         )
 
     @pytest.mark.it(
@@ -221,6 +230,7 @@ def test_configures_mqtt_websockets(self, mocker):
 
         assert mock_mqtt_client_constructor.call_count == 1
         assert mock_mqtt_client_constructor.call_args == mocker.call(
+            callback_api_version=mqtt.CallbackAPIVersion.VERSION1,
             client_id=fake_device_id,
             clean_session=False,
             protocol=mqtt.MQTTv311,
@@ -802,8 +812,11 @@ def test_client_raises_base_exception(
     @pytest.mark.it("Raises a custom Exception if Paho disconnect returns a failing rc code")
     @pytest.mark.parametrize(
         "error_params",
-        operation_return_codes,
-        ids=["{}->{}".format(x["name"], x["error"].__name__) for x in operation_return_codes],
+        disconnect_operation_return_codes,
+        ids=[
+            "{}->{}".format(x["name"], x["error"].__name__)
+            for x in disconnect_operation_return_codes
+        ],
     )
     def test_client_returns_failing_rc_code(
         self, mocker, mock_mqtt_client, transport, error_params