This is an automated email from the ASF dual-hosted git repository.
eladkal 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 52133f167b0 Allow opting out of forwarding Dag-level parameters in
DatabricksRunNowOperator (#70130)
52133f167b0 is described below
commit 52133f167b058037e499dd0045ec3f6bf6a7bfb5
Author: Arnav <[email protected]>
AuthorDate: Thu Aug 6 10:39:44 2026 +0530
Allow opting out of forwarding Dag-level parameters in
DatabricksRunNowOperator (#70130)
* Allow opting out of forwarding Dag-level parameters in
DatabricksRunNowOperator
The DatabricksRunNowOperator always forwards Dag-level parameters as
job_parameters, but some Databricks jobs do not expect or accept these
parameters, causing task failures. This change introduces a forward_dag_params
boolean parameter (defaulting to True) to allow opting out. Additionally, this
fixes a Windows-specific encoding issue when parsing imports in static checks,
guards the os.register_at_fork calls to prevent startup crashes on Windows, and
adds a fcntl mock inside conftes [...]
* Skip Databricks job_parameters auto-injection when conflicting legacy
parameter slots are used
The Databricks API run-now endpoint rejects job_parameters when combined
with notebook_params, python_params, jar_params, spark_submit_params,
python_named_params, or dbt_commands. This update automatically skips
forwarding DAG-level params into job_parameters if any of those conflicting
slots are present in the payload. Also updates docstrings, operator RST
documentation, and unit tests.
---
providers/databricks/docs/operators/run_now.rst | 9 +++-
.../providers/databricks/operators/databricks.py | 27 +++++++++-
.../unit/databricks/operators/test_databricks.py | 58 +++++++++++++++++++++-
3 files changed, 90 insertions(+), 4 deletions(-)
diff --git a/providers/databricks/docs/operators/run_now.rst
b/providers/databricks/docs/operators/run_now.rst
index 00ee6ed0fec..156049444e6 100644
--- a/providers/databricks/docs/operators/run_now.rst
+++ b/providers/databricks/docs/operators/run_now.rst
@@ -60,7 +60,14 @@ for this run.
If ``job_parameters`` is not set in ``json`` and the operator's ``params``
dict is
non-empty, ``params`` is forwarded as ``job_parameters`` as-is, so Airflow Dag
params can
be passed dynamically to a run without hardcoding them in ``json``. If
``json`` already
-contains ``job_parameters``, it is left untouched.
+contains ``job_parameters``, it is left untouched. You can set
``forward_dag_params=False`` to
+disable this parameter forwarding behavior.
+
+.. note::
+ The Databricks API does not permit ``job_parameters`` to be used in
combination with
+ ``notebook_params``, ``python_params``, ``jar_params``,
``spark_submit_params``,
+ ``python_named_params``, or ``dbt_commands``. Auto-forwarding is
automatically skipped
+ when any of those parameter slots are used.
.. code-block:: python
diff --git
a/providers/databricks/src/airflow/providers/databricks/operators/databricks.py
b/providers/databricks/src/airflow/providers/databricks/operators/databricks.py
index 2d17a21587f..f6bf1cf6aca 100644
---
a/providers/databricks/src/airflow/providers/databricks/operators/databricks.py
+++
b/providers/databricks/src/airflow/providers/databricks/operators/databricks.py
@@ -303,6 +303,16 @@ _DICT_PARAM_FIELD_BY_TASK = {
"run_job_task": "job_parameters",
}
+# Parameter slots in run-now payload that Databricks API rejects when combined
with job_parameters
+_RUN_NOW_PARAM_SLOTS_CONFLICTING_WITH_JOB_PARAMETERS = (
+ "notebook_params",
+ "python_params",
+ "jar_params",
+ "spark_submit_params",
+ "python_named_params",
+ "dbt_commands",
+)
+
def _inject_airflow_params_into_task(task: dict, params: dict) -> None:
"""Set dict-shaped per-task parameter fields from ``params`` if they are
not already set."""
@@ -1173,12 +1183,18 @@ class DatabricksRunNowOperator(ResumableJobMixin,
BaseOperator):
before polling begins so that a worker crash and retry reconnects to
the existing run
instead of triggering a duplicate run of the same job. Set to
``False`` to always trigger a
fresh run on retry. Requires Airflow 3.3+; on earlier versions it is
silently ignored.
+ :param forward_dag_params: Whether to forward Dag-level params as
``job_parameters``
+ when no ``job_parameters`` are specified. (default: ``True``)
.. note::
If ``job_parameters`` is not set in ``json`` and the operator's
``params`` dict is
non-empty, the operator's ``params`` are automatically forwarded as
``job_parameters``
so that Airflow Dag params can be passed dynamically to Databricks
runs without
- hardcoding them in ``json``.
+ hardcoding them in ``json``. Set ``forward_dag_params=False`` to
disable this.
+ Note that the Databricks API does not permit ``job_parameters`` to be
used in combination
+ with ``notebook_params``, ``python_params``, ``jar_params``,
``spark_submit_params``,
+ ``python_named_params``, or ``dbt_commands``; auto-forwarding is
automatically skipped
+ when any of those parameters are set.
"""
external_id_key = "databricks_run_now_id"
@@ -1229,6 +1245,7 @@ class DatabricksRunNowOperator(ResumableJobMixin,
BaseOperator):
repair_run: bool = False,
databricks_repair_reason_new_settings: dict[str, Any] | None = None,
cancel_previous_runs: bool = False,
+ forward_dag_params: bool = True,
**kwargs,
) -> None:
"""Create a new ``DatabricksRunNowOperator``."""
@@ -1254,6 +1271,7 @@ class DatabricksRunNowOperator(ResumableJobMixin,
BaseOperator):
self.repair_run = repair_run
self.databricks_repair_reason_new_settings =
databricks_repair_reason_new_settings or {}
self.cancel_previous_runs = cancel_previous_runs
+ self.forward_dag_params = forward_dag_params
# This variable will be used in case our task gets killed.
self.run_id: int | None = None
@@ -1321,7 +1339,12 @@ class DatabricksRunNowOperator(ResumableJobMixin,
BaseOperator):
json["job_id"] = job_id
del json["job_name"]
- if not json.get("job_parameters") and self.params:
+ if (
+ self.forward_dag_params
+ and not json.get("job_parameters")
+ and self.params
+ and not any(k in json for k in
_RUN_NOW_PARAM_SLOTS_CONFLICTING_WITH_JOB_PARAMETERS)
+ ):
json["job_parameters"] = dict(self.params)
return json
diff --git
a/providers/databricks/tests/unit/databricks/operators/test_databricks.py
b/providers/databricks/tests/unit/databricks/operators/test_databricks.py
index 9c83c19ebe9..ad9d718e34f 100644
--- a/providers/databricks/tests/unit/databricks/operators/test_databricks.py
+++ b/providers/databricks/tests/unit/databricks/operators/test_databricks.py
@@ -2028,7 +2028,7 @@ class TestDatabricksRunNowOperator:
durable=False,
task_id=TASK_ID,
job_id=JOB_ID,
- json={"notebook_params": {"a": "b"}},
+ json={"idempotency_token": "token_123"},
params={"env": "prod"},
)
op.render_template_fields(context={"ds": DATE})
@@ -2925,6 +2925,62 @@ class TestDatabricksRunNowOperator:
actual = db_mock.run_now.call_args.args[0]
assert actual["job_parameters"] == {"explicit": "value"}
+
@mock.patch("airflow.providers.databricks.operators.databricks.DatabricksHook")
+ def
test_run_now_does_not_inject_airflow_params_when_forward_dag_params_is_false(self,
db_mock_class):
+ """
+ When ``forward_dag_params`` is False, the operator's ``params`` should
+ not be forwarded as ``job_parameters``.
+ """
+ op = DatabricksRunNowOperator(
+ durable=False,
+ task_id=TASK_ID,
+ job_id=JOB_ID,
+ forward_dag_params=False,
+ params={"env": "prod", "batch_size": 100},
+ )
+ db_mock = db_mock_class.return_value
+ db_mock.run_now.return_value = RUN_ID
+ db_mock.get_run = make_run_with_state_mock("TERMINATED", "SUCCESS")
+
+ op.execute(None)
+
+ actual = db_mock.run_now.call_args.args[0]
+ assert "job_parameters" not in actual
+
+ @pytest.mark.parametrize(
+ ("slot_param", "slot_val"),
+ [
+ ("notebook_params", {"foo": "bar"}),
+ ("python_params", ["foo", "bar"]),
+ ("jar_params", ["foo", "bar"]),
+ ("spark_submit_params", ["--class", "Foo"]),
+ ("python_named_params", {"foo": "bar"}),
+ ("dbt_commands", ["dbt deps", "dbt run"]),
+ ],
+ )
+
@mock.patch("airflow.providers.databricks.operators.databricks.DatabricksHook")
+ def test_run_now_skips_param_injection_with_legacy_param_slots(self,
db_mock_class, slot_param, slot_val):
+ """
+ When any legacy param slot (notebook_params, etc.) is set,
auto-injection of ``self.params``
+ into ``job_parameters`` must be skipped because the Databricks API
rejects combining them.
+ """
+ op = DatabricksRunNowOperator(
+ durable=False,
+ task_id=TASK_ID,
+ job_id=JOB_ID,
+ params={"env": "prod"},
+ **{slot_param: slot_val},
+ )
+ db_mock = db_mock_class.return_value
+ db_mock.run_now.return_value = RUN_ID
+ db_mock.get_run = make_run_with_state_mock("TERMINATED", "SUCCESS")
+
+ op.execute(None)
+
+ actual = db_mock.run_now.call_args.args[0]
+ assert slot_param in actual
+ assert "job_parameters" not in actual
+
@pytest.mark.skipif(
not AIRFLOW_V_3_3_PLUS, reason="task_state_store (durable execution)
requires Airflow 3.3+"
