Providers: Allow opting out of forwarding Dag-level parameters in DatabricksRunNowOperator - #70130
Providers: Allow opting out of forwarding Dag-level parameters in DatabricksRunNowOperator#70130onlyarnav wants to merge 3 commits into
Conversation
jroachgolf84
left a comment
There was a problem hiding this comment.
Please review my comments, specifically, there are a few changes in there I don't think should be in there.
amoghrajesh
left a comment
There was a problem hiding this comment.
@onlyarnav there is some unrelated changes in your PR diff, please revise it and keep the changes only scoped to this PR.
…owOperator 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 conftest to enable the local test suite on Windows.
6d0b519 to
340211a
Compare
jroachgolf84
left a comment
There was a problem hiding this comment.
Thanks for the changes, LGTM.
|
cc @moomindani for a review from Databricks team |
moomindani
left a comment
There was a problem hiding this comment.
Thanks for the fix, and thanks for cleaning up the unrelated Windows changes — the diff is nicely scoped now.
I validated this against a real Databricks workspace, and the regression is more severe than the issue describes. The run-now endpoint rejects job_parameters outright when combined with any of the legacy param slots. All six confirmed:
job_parameters can not be used in combination with notebook_params
job_parameters can not be used in combination with python_params
job_parameters can not be used in combination with jar_params
job_parameters can not be used in combination with spark_submit_params
job_parameters can not be used in combination with python_named_params
job_parameters can not be used in combination with dbt_commands
Since Dag-level params propagate to the operator implicitly, a Dag like this has been failing since 7.16.0 with no user action:
with DAG("d", params={"env": "prod"}):
DatabricksRunNowOperator(task_id="t", job_id=1, notebook_params={"foo": "bar"})
# payload -> {"job_id": 1, "notebook_params": {...}, "job_parameters": {"env": "prod"}} # API rejectsI also checked whether an escape hatch already existed: job_parameters={} does not work, because not json.get("job_parameters") treats an empty dict as falsy and injects params anyway. So the new flag is genuinely necessary — the approach is right.
Verification I ran:
- Reverted the guard in
_build_run_now_payload→ the new test fails. Restored → full operator suite passes (201 tests). prek run --stage pre-commitclean on the diff.
The direction is good. I left three inline notes — the first one is the substantive one (default behaviour still leaves existing users broken); the others are the docstring and test coverage.
One more docs point I could not anchor inline, because the file is not part of this diff: providers/databricks/docs/operators/run_now.rst (the "Forwarding Airflow Dag params as Databricks job parameters" section, around line 60) still documents forwarding as unconditional. That guide is where users land when they hit this failure, so the opt-out should be discoverable there too — a sentence naming forward_dag_params=False, plus the mutual-exclusivity constraint with the legacy param slots.
Non-blocking, out of scope for this PR: #66613 added the same forwarding to DatabricksCreateJobsOperator (parameters) and DatabricksSubmitRunOperator (per-task dict slots), which have no opt-out. Scoping this to RunNow matches the issue, so no change needed here — just flagging it in case the flag should be applied consistently in a follow-up.
Drafted-by: Claude Code (Opus 5)
| 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: |
There was a problem hiding this comment.
Consider also skipping injection when a conflicting slot is already present.
Because forward_dag_params defaults to True, every Dag that is currently broken by #66613 stays broken until its author discovers this new flag — and the failure mode gives no hint that a new operator argument is the fix. The conflict is deterministic and fully knowable from the payload at this point, so it can be repaired automatically:
_RUN_NOW_PARAM_SLOTS_CONFLICTING_WITH_JOB_PARAMETERS = (
"notebook_params",
"python_params",
"jar_params",
"spark_submit_params",
"python_named_params",
"dbt_commands",
)
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)That fixes the regression for existing users without requiring them to change anything, and forward_dag_params=False then covers the remaining case the issue describes — a job whose entry point rejects extra params even when there is no slot conflict.
I'm happy to be argued out of this if maintainers prefer to keep the change minimal and purely opt-in; the silent-failure-by-default aspect is what makes me raise it.
There was a problem hiding this comment.
I've updated _build_run_now_payload to define _RUN_NOW_PARAM_SLOTS_CONFLICTING_WITH_JOB_PARAMETERS and automatically skip auto-injecting job_parameters whenever any of the 6 legacy parameter slots are present in json
| 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`` |
There was a problem hiding this comment.
The .. note:: a few lines below still states forwarding happens unconditionally, so the two now contradict each other. Worth adding the opt-out to the note, e.g. append: "Set forward_dag_params=False to disable this."
Also worth noting there that job_parameters cannot be combined with notebook_params / python_params / jar_params / spark_submit_params / python_named_params / dbt_commands — the API rejects such a run, which is the main reason a user would reach for this flag.
There was a problem hiding this comment.
Updated! I've updated the .. note:: in the DatabricksRunNowOperator docstring as well as the user documentation in providers/databricks/docs/operators/run_now.rst to explain forward_dag_params=False and document the Databricks API conflict with the legacy parameter slots
| 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): |
There was a problem hiding this comment.
The test correctly covers the flag's mechanics (I verified it fails without the guard). What isn't covered is the scenario that actually motivates the change: Dag-level params combined with one of the legacy param slots, which is what the API rejects. Nothing in the existing suite exercises that combination.
Worth adding a case asserting that with notebook_params set and forward_dag_params=False, the payload contains notebook_params and no job_parameters — that pins the regression scenario rather than just the flag. If you take the auto-skip suggestion on _build_run_now_payload, the same case parametrized over the six conflicting slots would cover it directly.
There was a problem hiding this comment.
Done! I've added a parametrized test test_run_now_skips_param_injection_with_legacy_param_slots covering all 6 conflicting parameter slots to verify that job_parameters is not injected when any of them are present
… 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.
fa593d3 to
90e726b
Compare
moomindani
left a comment
There was a problem hiding this comment.
All three addressed, verified by running 90e726b6 rather than reading the summary. The auto-skip is implemented as suggested, and the four paths behave correctly:
Dag params + notebook_params -> {job_id, notebook_params} # regression repaired, no user change
explicit job_parameters + slot -> both preserved # user intent respected
Dag params, no slot -> {job_id, job_parameters} # forwarding still works
forward_dag_params=False -> {job_id} # flag still honoured
Keeping an explicitly-passed job_parameters rather than silently dropping it is the right call — only the implicit forwarding is skipped, so a user who deliberately combined them still gets the API error instead of a silent behaviour change. 207 tests pass, and the test_execute_does_not_mutate_json_template_field edit is a necessary consequence of the skip logic with its assertions intact.
I went back to the workspace to check the approach itself, and found the constraint is bidirectional — worth knowing, though it does not change my assessment:
| Job config | Sent | Result |
|---|---|---|
has job-level parameters |
notebook_params only |
rejected: "Cannot use legacy parameters (...) because the job has job parameters configured" |
has job-level parameters |
job_parameters only |
accepted |
no job-level parameters |
notebook_params only |
accepted |
| either | both | rejected: "job_parameters can not be used in combination with notebook_params" |
So for a job that does declare job-level parameters, skipping the injection trades one rejection for the other. But that combination was already impossible before #66613, so the set of broken Dags is unchanged — and the case your fix actually repairs (job without job-level parameters + legacy slot + Dag params) is a genuine #66613 regression. The skip condition maps correctly onto the API constraint, and forward_dag_params still earns its place for jobs whose entry point rejects extra params even with no slot conflict. Approach looks right to me.
One optional docs nit, non-blocking: the new note says job_parameters cannot be combined with the legacy slots, which is accurate but narrower than the API's actual rule. A reader may conclude "then I'll just use notebook_params" — which also fails if the job declares job-level parameters. A clause like "and the legacy slots cannot be used at all against a job that declares job-level parameters" would close that gap. Fine to leave for a follow-up.
Drafted-by: Claude Code (Opus 5)
Description
This PR introduces an option to opt out of forwarding Dag-level parameters in the
DatabricksRunNowOperator.In PR #66613 (for issue #39002), parameter forwarding was introduced such that the operator's
paramsdict is automatically forwarded asjob_parameterswhen nojob_parametersare specified. However, this changes the payload for Databricks jobs and breaks execution for jobs whose entry points do not expect or accept additional parameters.To resolve this compatibility issue, this PR:
forward_dag_params(default:True) toDatabricksRunNowOperator.self.forward_dag_paramswhen constructing the payload inside_build_run_now_payloadbefore mergingself.params.forward_dag_params=False.os.register_at_fork, and a mockfcntl) to enable executing the local test suite on Windows development hosts.closes: #70121
Was generative AI tooling used to co-author this PR?
Generated-by: Google Antigravity following the guidelines