gaurav7261 commented on code in PR #39110:
URL: https://github.com/apache/airflow/pull/39110#discussion_r1573321766
##########
airflow/providers/databricks/operators/databricks.py:
##########
@@ -156,17 +144,37 @@ def
_handle_deferrable_databricks_operator_execution(operator, hook, log, contex
log.info("%s completed successfully.", operator.task_id)
-def _handle_deferrable_databricks_operator_completion(event: dict, log:
Logger) -> None:
+def _get_databricks_notebook_error(run_info: dict, hook: DatabricksHook,
run_state: RunState) -> str:
+ task_run_id = None
+ if "tasks" in run_info:
+ for task in run_info["tasks"]:
+ if task.get("state", {}).get("result_state", "") == "FAILED":
+ task_run_id = task["run_id"]
+ if task_run_id is not None:
+ run_output = hook.get_run_output(task_run_id)
+ if "error" in run_output:
+ notebook_error = run_output["error"]
+ else:
+ notebook_error = run_state.state_message
+ else:
+ notebook_error = run_state.state_message
+ return notebook_error
+
+
+def _handle_deferrable_databricks_operator_completion(event: dict, log:
Logger, hook: DatabricksHook) -> None:
validate_trigger_event(event)
run_state = RunState.from_json(event["run_state"])
run_page_url = event["run_page_url"]
+ run_id = event["run_id"]
log.info("View run status, Spark UI, and logs at %s", run_page_url)
if run_state.is_successful:
log.info("Job run completed successfully.")
return
+ run_info = hook.get_run(run_id)
Review Comment:
sure thanks for feedback, working on it
--
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]
