(airflow) branch main updated: Fix TaskSDK swallowing errors when Variable.set() or Variable.delete() fails (#68542)

[ash]

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

ashb 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 c59f4cd8bb7 Fix TaskSDK swallowing errors when Variable.set() or 
Variable.delete() fails (#68542)
c59f4cd8bb7 is described below

commit c59f4cd8bb7123f3493a41995b22f83db768f0bf
Author: Jayachandra Kasarla <jayachandra.kasa...@astronomer.io>
AuthorDate: Thu Jun 18 17:41:53 2026 +0530

    Fix TaskSDK swallowing errors when Variable.set() or Variable.delete() 
fails (#68542)
    
    It was being logged, but no error was returned to the caller.
---
 task-sdk/src/airflow/sdk/definitions/variable.py      | 12 ++----------
 task-sdk/tests/task_sdk/definitions/test_variables.py | 17 ++++++++++++++---
 2 files changed, 16 insertions(+), 13 deletions(-)

diff --git a/task-sdk/src/airflow/sdk/definitions/variable.py 
b/task-sdk/src/airflow/sdk/definitions/variable.py
index a379022a909..f1db76c3fae 100644
--- a/task-sdk/src/airflow/sdk/definitions/variable.py
+++ b/task-sdk/src/airflow/sdk/definitions/variable.py
@@ -60,13 +60,9 @@ class Variable:
 
     @classmethod
     def set(cls, key: str, value: Any, description: str | None = None, 
serialize_json: bool = False) -> None:
-        from airflow.sdk.exceptions import AirflowRuntimeError
         from airflow.sdk.execution_time.context import _set_variable
 
-        try:
-            return _set_variable(key, value, description, 
serialize_json=serialize_json)
-        except AirflowRuntimeError as e:
-            log.exception(e)
+        _set_variable(key, value, description, serialize_json=serialize_json)
 
     @classmethod
     def keys(cls, prefix: str | None = None) -> Sequence[str]:
@@ -94,10 +90,6 @@ class Variable:
 
     @classmethod
     def delete(cls, key: str) -> None:
-        from airflow.sdk.exceptions import AirflowRuntimeError
         from airflow.sdk.execution_time.context import _delete_variable
 
-        try:
-            _delete_variable(key=key)
-        except AirflowRuntimeError as e:
-            log.exception(e)
+        _delete_variable(key=key)
diff --git a/task-sdk/tests/task_sdk/definitions/test_variables.py 
b/task-sdk/tests/task_sdk/definitions/test_variables.py
index 6e94ccf503f..29b6ac0cb97 100644
--- a/task-sdk/tests/task_sdk/definitions/test_variables.py
+++ b/task-sdk/tests/task_sdk/definitions/test_variables.py
@@ -25,7 +25,15 @@ import pytest
 
 from airflow.sdk import Variable
 from airflow.sdk.configuration import initialize_secrets_backends
-from airflow.sdk.execution_time.comms import GetVariableKeys, PutVariable, 
VariableKeysResult, VariableResult
+from airflow.sdk.exceptions import AirflowRuntimeError, ErrorType
+from airflow.sdk.execution_time.comms import (
+    DeleteVariable,
+    ErrorResponse,
+    GetVariableKeys,
+    PutVariable,
+    VariableKeysResult,
+    VariableResult,
+)
 from airflow.sdk.execution_time.secrets import 
DEFAULT_SECRETS_SEARCH_PATH_WORKERS
 
 from tests_common.test_utils.config import conf_vars
@@ -89,6 +97,11 @@ class TestVariables:
             ),
         )
 
+    def test_var_delete(self, mock_supervisor_comms):
+        Variable.delete(key="my_key")
+
+        
mock_supervisor_comms.send.assert_called_once_with(msg=DeleteVariable(key="my_key"))
+
 
 class TestVariableKeys:
     @pytest.mark.parametrize(
@@ -171,8 +184,6 @@ class TestVariableKeys:
         )
 
     def test_keys_raises_on_error_response(self, mock_supervisor_comms):
-        from airflow.sdk.exceptions import AirflowRuntimeError, ErrorType
-        from airflow.sdk.execution_time.comms import ErrorResponse
 
         mock_supervisor_comms.send.return_value = ErrorResponse(
             error=ErrorType.GENERIC_ERROR, detail={"message": "boom"}