amoghrajesh commented on code in PR #76718:
URL: https://github.com/apache/airflow/pull/76718#discussion_r4520474482
##########
providers/databricks/tests/unit/databricks/operators/test_databricks.py:
##########
@@ -2229,7 +2262,7 @@ def test_resubmits_when_stored_run_no_longer_exists(self,
db_mock_class):
db_mock = db_mock_class.return_value
db_mock.submit_run.return_value = RUN_ID
db_mock.get_run.side_effect = [
- DatabricksApiError("Response: RESOURCE_DOES_NOT_EXIST, Status
Code: 404", http_status_code=404),
+ make_api_error(400, "INVALID_PARAMETER_VALUE", "Run 999 does not
exist."),
Review Comment:
[final review from
kaxil](https://github.com/apache/airflow/pull/76718/commits/9a29f65bba08a4b712fed52ede265e5d0df4a10c)
##########
providers/databricks/src/airflow/providers/databricks/operators/databricks.py:
##########
@@ -82,6 +83,19 @@
XCOM_RUN_PAGE_URL_KEY = "run_page_url"
XCOM_STATEMENT_ID_KEY = "statement_id"
+# jobs/runs/get answers 400 INVALID_PARAMETER_VALUE for a run it no longer
knows about, never 404.
+# These are the two message forms the Databricks SDK overrides to
ResourceDoesNotExist; the code
+# alone also covers malformed input, which must stay an error rather than
trigger a resubmit.
+_RUN_MISSING_MESSAGE_PATTERN = re.compile(r"Run .* does not exist|Run: .* in
job: .* doesn't exist")
+
+
+def is_run_missing_error(exc: DatabricksApiError) -> bool:
+ """Whether ``exc`` means Databricks has no record of the run (expired,
deleted, or foreign)."""
+ if exc.http_status_code == 404 or exc.error_code ==
"RESOURCE_DOES_NOT_EXIST":
Review Comment:
[final review from
kaxil](https://github.com/apache/airflow/pull/76718/commits/9a29f65bba08a4b712fed52ede265e5d0df4a10c)
##########
providers/databricks/src/airflow/providers/databricks/operators/databricks.py:
##########
@@ -82,6 +83,19 @@
XCOM_RUN_PAGE_URL_KEY = "run_page_url"
XCOM_STATEMENT_ID_KEY = "statement_id"
+# jobs/runs/get answers 400 INVALID_PARAMETER_VALUE for a run it no longer
knows about, never 404.
+# These are the two message forms the Databricks SDK overrides to
ResourceDoesNotExist; the code
+# alone also covers malformed input, which must stay an error rather than
trigger a resubmit.
+_RUN_MISSING_MESSAGE_PATTERN = re.compile(r"Run .* does not exist|Run: .* in
job: .* doesn't exist")
+
+
+def is_run_missing_error(exc: DatabricksApiError) -> bool:
Review Comment:
[final review from
kaxil](https://github.com/apache/airflow/pull/76718/commits/9a29f65bba08a4b712fed52ede265e5d0df4a10c)
##########
providers/databricks/src/airflow/providers/databricks/operators/databricks.py:
##########
@@ -1024,9 +1038,8 @@ def get_job_status(self, external_id: JsonValue, context:
Context) -> str:
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:
+ # Neither active nor succeeded, so the mixin resubmits instead of
failing every retry.
+ if is_run_missing_error(e):
Review Comment:
[final review from
kaxil](https://github.com/apache/airflow/pull/76718/commits/9a29f65bba08a4b712fed52ede265e5d0df4a10c)
--
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]
