kaxil commented on code in PR #68974:
URL: https://github.com/apache/airflow/pull/68974#discussion_r3474712130
##########
providers/databricks/src/airflow/providers/databricks/operators/databricks.py:
##########
@@ -876,6 +909,46 @@ def
_inject_openlineage_properties_into_databricks_job(self, json: dict, context
)
return json
+ def submit_job(self, context: Context) -> int:
+ normalised = self._prepare_submit_json(context)
+ # 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.submit_run(normalised)
+ self.log.info("Run submitted with run_id: %s", self.run_id)
+ return self.run_id
+
+ def get_job_status(self, external_id: int, 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.
+ run_info = self._hook.get_run(external_id)
Review Comment:
On a retry, if the stored run no longer exists on Databricks (run history
expired, or the run was deleted), `get_run` raises a non-retryable HTTP error
and the whole task fails instead of falling back to a fresh submit. The
`SparkSubmitOperator` sibling handles this case explicitly: its
`get_job_status` maps a vanished job (K8s 404) to a `"NotFound"` status that's
neither active nor succeeded, so the mixin resubmits fresh. Worth catching the
not-found error here and returning a status that
`is_job_active`/`is_job_succeeded` both treat as false, so a lost run id
degrades to resubmit rather than a hard failure.
--
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]
