kaxil commented on code in PR #69174:
URL: https://github.com/apache/airflow/pull/69174#discussion_r3501688382
##########
providers/databricks/src/airflow/providers/databricks/operators/databricks.py:
##########
@@ -1291,23 +1308,75 @@ def execute(self, context: Context):
json["job_id"] = job_id
del json["job_name"]
+ if not json.get("job_parameters") and self.params:
+ json["job_parameters"] = dict(self.params)
+
+ return json
+
+ def _prepare_run_now_json(self) -> dict[str, Any]:
+ json = self._build_run_now_payload()
if self.cancel_previous_runs:
if (job_id := json.get("job_id")) is None:
raise ValueError(
"cancel_previous_runs=True requires either job_id or
job_name to be provided."
)
+ self._hook.cancel_all_runs(job_id)
- hook.cancel_all_runs(job_id)
+ self._merged_json = json
+ return json
- if not json.get("job_parameters") and self.params:
- json["job_parameters"] = dict(self.params)
+ def submit_job(self, context: Context) -> int:
+ json = self._prepare_run_now_json()
+ # Set run_id the instant the run exists so on_kill can cancel it even
if the worker dies
+ # before polling begins.
+ self.run_id = self._hook.run_now(json)
+ self.log.info("Run submitted with run_id: %s", self.run_id)
+ return self.run_id
- self._merged_json = json
- self.run_id = hook.run_now(json)
- if self.deferrable:
- _handle_deferrable_databricks_operator_execution(self, hook,
self.log, context)
- else:
- _handle_databricks_operator_execution(self, hook, self.log,
context)
+ def get_job_status(self, external_id: JsonValue, context: Context) -> str:
+ # Databricks splits run state across life_cycle_state and
result_state; the mixin's status
+ # interface is a single string, so encode both as "LIFE_CYCLE:RESULT"
and decode them in
+ # is_job_active / is_job_succeeded.
+ try:
+ run_info = self._hook.get_run(external_id)
+ except DatabricksApiError as e:
+ # A stored run whose history expired (or was deleted) returns 404.
Report a sentinel that
+ # is neither active nor succeeded so the mixin resubmits fresh
instead of failing the task.
+ if e.http_status_code == 404:
+ return "NOT_FOUND:"
+ raise
+ state = RunState(**run_info["state"])
Review Comment:
On a durable retry, if the stored run is currently `BLOCKED` or
`WAITING_FOR_RETRY`, this raises `AirflowException("Unexpected life cycle
state")` instead of reconnecting. Both states are defined in
`RunLifeCycleState` (`hooks/databricks.py:81,89`) but missing from
`RunState.RUN_LIFE_CYCLE_STATES` (`hooks/databricks.py:95-103`), and
`RunState.__init__` raises on anything outside that allowlist. It's an
`AirflowException`, not a `DatabricksApiError`, so the 404 fallback above
doesn't catch it and the task fails instead of waiting for a run that's still
in flight.
These are exactly the in-flight, not-yet-terminal states the reconnect path
exists to handle, and `RunNow` is more exposed than `SubmitRun` since it
triggers existing jobs that are often concurrency- or dependency-gated, so
`BLOCKED` is routine.
This isn't introduced here -- the same `RunState(**run_info["state"])`
strictness governs the normal poll loop
(`_handle_databricks_operator_execution`) and `get_run_state`, so a `BLOCKED`
run already fails on a clean run today. Adding the two states to
`is_job_active` wouldn't be enough, since the crash happens earlier at
`RunState(...)`; the fix belongs in `RunState.RUN_LIFE_CYCLE_STATES` (and
`is_terminal`) so it covers both operators and the happy path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
