Merge branch 'ct-paho-2' of https://github.com/Azure/azure-iot-sdk-python into ct-paho-2

Author: cartertinney

dev_utils/dev_utils/service_helper.py

@@ -5,6 +5,8 @@
 import asyncio
 import concurrent.futures
 
+# TODO: Consider removing injectable executors.
+
 
 class ServiceHelper:
     def __init__(
@@ -14,6 +16,7 @@ def __init__(
         eventhub_consumer_group,
         executor=None,
     ):
+        self._owns_executor = executor is None
         self._executor = executor or concurrent.futures.ThreadPoolExecutor()
         self._inner_object = ServiceHelperSync(
             iothub_connection_string, eventhub_connection_string, eventhub_consumer_group
@@ -66,4 +69,8 @@ async def get_next_reported_patch_arrival(self, block=True, timeout=240):
         )
 
     async def shutdown(self):
-        return await self._run_in_executor(self._inner_object.shutdown)
+        await self._run_in_executor(self._inner_object.shutdown)
+
+        if self._owns_executor:
+            self._executor.shutdown(wait=True)
+        return

dev_utils/dev_utils/service_helper_sync.py

@@ -201,6 +201,16 @@ def shutdown(self):
         if self._eventhub_consumer_client:
             self._eventhub_consumer_client.close()
 
+        if self._eventhub_future:
+            try:
+                # Ensure the EventHub receive loop exits before tearing down the executor
+                self._eventhub_future.result(timeout=30)
+            except Exception:
+                logger.warning("_eventhub_thread did not exit cleanly", exc_info=True)
+
+        if self._executor:
+            self._executor.shutdown(wait=True)
+
     def _convert_incoming_event(self, event):
         try:
             event_body = event.body_as_json()

tests/e2e/iothub_e2e/conftest.py

@@ -43,7 +43,9 @@ def transport():
 
 @pytest.fixture(scope="session")
 def executor():
-    return concurrent.futures.ThreadPoolExecutor()
+    executor = concurrent.futures.ThreadPoolExecutor()
+    yield executor
+    executor.shutdown()
 
 
 @pytest.fixture(scope="function")