Switch drain_all call to be called from new thread instead of event…

… loop. (#16739)

flows/setup_logging.py

@@ -0,0 +1,24 @@
+"""
+This is a regression tests for https://github.com/PrefectHQ/prefect/issues/16115
+"""
+
+import logging
+from logging.config import dictConfig
+
+from prefect import flow
+
+logger = logging.getLogger(__name__)
+logger.setLevel("WARNING")
+
+dictConfig({"version": 1, "incremental": False})
+
+
+@flow
+def myflow():
+    return "Hello"
+
+
+if __name__ == "__main__":
+    result = myflow()
+
+    assert result == "Hello"

src/prefect/logging/handlers.py

@@ -117,8 +117,8 @@ def flush(cls) -> None:
                 )
 
             # Not ideal, but this method is called by the stdlib and cannot return a
-            # coroutine so we just schedule the drain in the global loop thread and continue
-            from_sync.call_soon_in_loop_thread(create_call(APILogWorker.drain_all))
+            # coroutine so we just schedule the drain in a new thread and continue
+            from_sync.call_soon_in_new_thread(create_call(APILogWorker.drain_all))
             return None
         else:
             # We set a timeout of 5s because we don't want to block forever if the worker