Handle triggerer cross-loop connection fallback (#64213)

Author: deepujain

task-sdk/src/airflow/sdk/execution_time/secrets/execution_api.py

@@ -26,6 +26,12 @@
     from airflow.sdk import Connection
 
 
+_GREENBACK_RUNTIME_ERROR_MESSAGES = (
+    "You cannot use AsyncToSync in the same thread as an async event loop",
+    "got Future <Future pending> attached to a different loop",
+)
+
+
 class ExecutionAPISecretsBackend(BaseSecretsBackend):
     """
     Secrets backend for client contexts (workers, DAG processors, triggerers).
@@ -69,7 +75,7 @@ def get_connection(self, conn_id: str, team_name: str | None = None) -> Connecti
             # TriggerCommsDecoder.send() uses async_to_sync internally, which raises RuntimeError
             # when called within an async event loop. In greenback portal contexts (triggerer),
             # we catch this and use greenback to call the async version instead.
-            if str(e).startswith("You cannot use AsyncToSync in the same thread as an async event loop"):
+            if any(message in str(e) for message in _GREENBACK_RUNTIME_ERROR_MESSAGES):
                 import asyncio
 
                 import greenback

task-sdk/tests/task_sdk/execution_time/test_secrets.py

@@ -121,12 +121,22 @@ def test_get_conn_value_not_implemented(self):
         with pytest.raises(NotImplementedError, match="Use get_connection instead"):
             backend.get_conn_value("test_conn")
 
-    def test_runtime_error_triggers_greenback_fallback(self, mocker, mock_supervisor_comms):
+    @pytest.mark.parametrize(
+        "runtime_error_message",
+        [
+            "You cannot use AsyncToSync in the same thread as an async event loop",
+            "Task <Task pending name='Task-3'> got Future <Future pending> attached to a different loop",
+        ],
+    )
+    def test_runtime_error_triggers_greenback_fallback(
+        self, mocker, mock_supervisor_comms, runtime_error_message
+    ):
         """
         Test that RuntimeError from async_to_sync triggers greenback fallback.
 
-        This test verifies the fix for issue #57145: when SUPERVISOR_COMMS.send()
-        raises the specific RuntimeError about async_to_sync in an event loop,
+        This test verifies the fix for issue #57145 and regression coverage for #64213:
+        when SUPERVISOR_COMMS.send() raises the specific RuntimeError about async_to_sync
+        in an event loop, or the cross-loop Future error seen in triggerer worker threads,
         the backend catches it and uses greenback to call aget_connection().
         """
 
@@ -138,9 +148,7 @@ def test_runtime_error_triggers_greenback_fallback(self, mocker, mock_supervisor
         )
 
         # Simulate the RuntimeError that triggers greenback fallback
-        mock_supervisor_comms.send.side_effect = RuntimeError(
-            "You cannot use AsyncToSync in the same thread as an async event loop"
-        )
+        mock_supervisor_comms.send.side_effect = RuntimeError(runtime_error_message)
 
         # Mock the greenback and asyncio modules that are imported inside the exception handler
         mocker.patch("greenback.has_portal", return_value=True)