Task SDK: Fix Variable.set/delete swallowing API errors — propagate AirflowRuntimeError to fail the task

[Shushankranjan]

When the Execution API rejects a variable write (any non-2xx response —
e.g. a 403 authorization denial or a server error), Variable.set() and
Variable.delete() were catching AirflowRuntimeError and only logging
it. The task still finished as success and the variable was left
unchanged.

This is inconsistent with Variable.get(), which already raises on error
(and since #66575 even refuses secrets-backend fallback on a 401/403).
Reads fail loudly on denial; writes silently succeeded.

Changes

• Variable.set(): Remove the try/except AirflowRuntimeError block
  that swallowed the error. Any API rejection now propagates and fails the
  task.
• Variable.delete(): Same fix.
• Remove the now-unused logging import and log variable from
  variable.py.
• Add test_var_set_raises_on_error and test_var_delete_raises_on_error
  to cover the error propagation path.

Root cause

# Before (broken)
try:
    _set_variable(...)
except AirflowRuntimeError as e:
    log.exception(e)   # swallowed — task reports success

# After (fixed)
_set_variable(...)     # AirflowRuntimeError propagates — task fails

closes: #68537
Related: #66575 (fixed the analogous read / secrets-backend path)

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example Dag that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[onlyarnav]

Could you explain the removal of logging and AirflowRuntimeError workflow from variable.py?

[Shushankranjan]

What Changed in variable.py

The change (from your previous conversation) was specifically in Variable.set() and Variable.delete(). Here's the before/after:

Variable.set() - Before

@classmethod
def set(cls, key, value, description=None, serialize_json=False) -> None:
    import logging
    from airflow.sdk.execution_time.context import _set_variable

    try:
        _set_variable(key, value, description, serialize_json=serialize_json)
    except AirflowRuntimeError:
        logging.getLogger(__name__).warning("Failed to set variable %r", key)
        # error was swallowed — task continued silently

Variable.set() - After (current)

@classmethod
def set(cls, key, value, description=None, serialize_json=False) -> None:
    from airflow.sdk.execution_time.context import _set_variable

    _set_variable(key, value, description, serialize_json=serialize_json)

The same pattern applied to Variable.delete().

---

Why the Change Was Made

1. Silent Failure is Dangerous

The old try/except caught and suppressed AirflowRuntimeError from the Execution API. If the API server rejected a set() or delete() (e.g., due to permissions, network issues, or server errors), the task would:

• Log a warning
• Continue executing as if the write succeeded ✗

This is a correctness bug: downstream tasks could read stale variable values or operate on data that was never actually written.

2. Inconsistency with Variable.get()

Variable.get() already let AirflowRuntimeError propagate (with one deliberate exception: VARIABLE_NOT_FOUND + a provided default). The old set()/delete() had no such intentional swallowing - it was just a side effect of defensive coding that went too far.

3. The try/except Was an Overcorrection

There's no legitimate reason to swallow a write error. The correct behavior is:

• Write fails → task fails → Airflow marks the task instance as failed → the user investigates

This matches Airflow's fault model: tasks are expected to be retried or alarmed on, not silently succeed.

---

What the Tests Verify Now

The two new tests confirm the corrected behavior:

Test	What it asserts
test_var_set_raises_on_error	Variable.set() re-raises AirflowRuntimeError when the API rejects the write
test_var_delete_raises_on_error	Variable.delete() re-raises AirflowRuntimeError when the API rejects the delete

The logging import was also removed as a cleanup - it was only there to service the now-gone except block.

---

Summary: The try/except + logging in set()/delete() was silently eating API errors, letting tasks proceed as if writes succeeded when they hadn't. Removing it makes write failures immediately visible as task failures, consistent with how get() already behaves.