(airflow) branch main updated: Handle supervisor remote log upload failures gracefully (#65308)

[ep...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

ephraimanierobi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/main by this push:
     new 6f1c97b40da Handle supervisor remote log upload failures gracefully (#65308)
6f1c97b40da is described below

commit 6f1c97b40da526c37293df86ed4fbf1f09ad9d94
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Wed Apr 15 15:04:33 2026 +0100

    Handle supervisor remote log upload failures gracefully (#65308)
    
    * Handle supervisor remote log upload failures gracefully
    
    Prevent remote log upload errors from crashing task supervisor shutdown.
    Catch upload exceptions in the supervisor, log the failure with task
    context, and keep task completion behavior intact.
    
    * Update task-sdk/src/airflow/sdk/execution_time/supervisor.py
    
    Co-authored-by: Ash Berlin-Taylor <as...@firemirror.com>
    
    * fixup! Update task-sdk/src/airflow/sdk/execution_time/supervisor.py
    
    ---------
    
    Co-authored-by: Ash Berlin-Taylor <as...@firemirror.com>
---
 .../src/airflow/sdk/execution_time/supervisor.py   |  7 ++++--
 .../task_sdk/execution_time/test_supervisor.py     | 25 ++++++++++++++++++++++
 2 files changed, 30 insertions(+), 2 deletions(-)

diff --git a/task-sdk/src/airflow/sdk/execution_time/supervisor.py b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
index 87e171d8cbc..4b27314e0eb 100644
--- a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
+++ b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
@@ -1097,8 +1097,11 @@ class ActivitySubprocess(WatchedSubprocess):
         """
         from airflow.sdk.log import upload_to_remote
 
-        with _remote_logging_conn(self.client):
-            upload_to_remote(self.process_log, self.ti)
+        try:
+            with _remote_logging_conn(self.client):
+                upload_to_remote(self.process_log, self.ti)
+        except Exception:
+            self.process_log.exception("Failed to upload remote logs", ti_id=self.id, pid=self.pid)
 
     def _monitor_subprocess(self):
         """
diff --git a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
index 39540913f45..3ab95e0810e 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
@@ -2915,6 +2915,31 @@ def test_remote_logging_conn(remote_logging, remote_conn, expected_env, monkeypa
                 assert connection_available["conn_uri"] is not None, "Connection URI was None during upload"
 
 
+def test_log_upload_failures_are_non_fatal(mocker):
+    proc = ActivitySubprocess(
+        process_log=mocker.MagicMock(),
+        id=TI_ID,
+        pid=12345,
+        stdin=mocker.MagicMock(),
+        client=mocker.MagicMock(),
+        process=mocker.MagicMock(),
+    )
+    proc.ti = mocker.MagicMock()
+
+    mocker.patch(
+        "airflow.sdk.execution_time.supervisor._remote_logging_conn",
+        side_effect=RuntimeError("upload failed"),
+    )
+
+    proc._upload_logs()
+
+    proc.process_log.exception.assert_called_once_with(
+        "Failed to upload remote logs",
+        ti_id=TI_ID,
+        pid=12345,
+    )
+
+
 def test_remote_logging_conn_sets_process_context(monkeypatch, mocker):
     """
     Test that _remote_logging_conn sets _AIRFLOW_PROCESS_CONTEXT=client.